diff --git a/.github/checks-manifest.yaml b/.github/checks-manifest.yaml index d9207298818..181a88d2e5a 100644 --- a/.github/checks-manifest.yaml +++ b/.github/checks-manifest.yaml @@ -4,6 +4,11 @@ # Checks that consume PR metadata declare `requires_pr_body: true`; post-merge # push runs exclude only that category because no authoritative PR body is available. checks: + - id: pcloud-backup-contract-tests + command: ["bash", "scripts/run-plugin-pcloud-tests.sh"] + triggers: ["plugins/omi-pcloud-app/**", "plugins/omi-plugin-sdk/**", "backend/utils/plugin_transport.py", "backend/utils/plugin_backup_audio.py", "scripts/run-plugin-pcloud-tests.sh", ".github/checks-manifest.yaml"] + lanes: ["local", "ci"] + reason: "#20498 passed unrelated checks while UID-only relinking, default-on private export and destructive upload retries escaped its direct-coroutine tests; exercise real installed HTTP routing, encrypted durable state, backend-issued grants and failure recovery" - id: speaker-identity-interpolation-ratchet command: ["python3", ".github/scripts/check_speaker_label_interpolation.py", "--changed-files", "{changed_files}", "--base", "{base}"] triggers: ["backend/**/*.py", "app/lib/**/*.dart", ".github/scripts/*speaker_label_interpolation.py", ".github/checks-manifest.yaml"] diff --git a/app/changelog/unreleased/20261020-app-home-fresh-launch-grant.json b/app/changelog/unreleased/20261020-app-home-fresh-launch-grant.json new file mode 100644 index 00000000000..3d079d5de63 --- /dev/null +++ b/app/changelog/unreleased/20261020-app-home-fresh-launch-grant.json @@ -0,0 +1,3 @@ +{ + "change": "Opening app settings or retrying an integration setup now uses a fresh sign-in link instead of an expired or already-used link" +} diff --git a/app/lib/backend/http/api/apps.dart b/app/lib/backend/http/api/apps.dart index 1cb6c5742a1..8c50a60163a 100644 --- a/app/lib/backend/http/api/apps.dart +++ b/app/lib/backend/http/api/apps.dart @@ -3,6 +3,8 @@ import 'dart:developer'; import 'dart:io'; import 'package:omi/backend/http/api_result.dart'; +import 'package:omi/backend/http/http_pool_manager.dart'; +import 'package:http/http.dart' as http; import 'package:omi/backend/http/shared.dart'; import 'package:omi/backend/preferences.dart'; import 'package:omi/backend/schema/app.dart'; @@ -408,26 +410,35 @@ String appSetupUrlWithUid(String url, String uid) { return '${head.substring(0, queryIndex)}?${[...kept, uidParam].join('&')}$fragment'; } -Future isAppSetupCompleted(String? url) async { +Future isAppSetupCompleted(String? url, {ApiSend? send, String? uid}) async { if (url == null || url.isEmpty) return true; - Logger.debug('isAppSetupCompleted: $url'); - var response = await makeApiCall( - url: appSetupUrlWithUid(url, SharedPreferencesUtil().uid), - method: 'GET', - headers: {}, - body: '', + final target = Uri.tryParse(url); + final omi = send == null ? Uri.tryParse(Env.apiBaseUrl ?? '') : null; + final isOmiOrigin = target != null && + omi != null && + omi.host.isNotEmpty && + target.scheme == omi.scheme && + target.host == omi.host && + target.port == omi.port; + // Plugin capabilities are the vendor authority. Never forward Omi's Firebase + // bearer token, refresh the Omi session on a vendor 401, or log signed URLs. + final ApiSend? transport = send ?? + (isOmiOrigin + ? null + : (request) => HttpPoolManager.instance.send(() => http.Request(request.method, Uri.parse(request.url)), + timeout: const Duration(seconds: 15), retries: 0)); + final account = uid ?? SharedPreferencesUtil().uid; + final result = await executeApi( + canSend: send == null ? () => SharedPreferencesUtil().uid == account : null, + request: ApiRequest(url: appSetupUrlWithUid(url, account), method: 'GET'), + send: transport, + decode: (body) { + final data = jsonDecode(body); + if (data is! Map || data['is_setup_completed'] is! bool) throw const FormatException('Invalid setup status'); + return data['is_setup_completed'] == true; + }, ); - try { - final data = jsonDecode(response?.body ?? '{}') as Map; - Logger.debug(data); - return data['is_setup_completed'] ?? false; - } on FormatException catch (e) { - Logger.debug('Response not a valid json: $e'); - return false; - } catch (e) { - Logger.debug('Error triggering request at endpoint: $e'); - return false; - } + return result is ApiSuccess && result.data; } Future<(bool, String, String?)> submitAppServer(File file, Map appData) async { @@ -573,13 +584,13 @@ Future?> getAppDetailsServer(String appId) async { var response = await makeApiCall(url: '${Env.apiBaseUrl}v1/apps/$appId', headers: {}, body: '', method: 'GET'); try { if (response == null || response.statusCode != 200) return null; - log('getAppDetailsServer: ${response.body}'); + Logger.debug('getAppDetailsServer: ${response.statusCode}'); final data = jsonDecode(response.body) as Map; wire.GeneratedApp.fromJson(data); return data; } catch (e, stackTrace) { - Logger.debug(e.toString()); - PlatformManager.instance.crashReporter.reportCrash(e, stackTrace); + Logger.debug('getAppDetailsServer decode failed: ${e.runtimeType}'); + PlatformManager.instance.crashReporter.reportCrash(StateError('App details decode failed'), stackTrace); return null; } } diff --git a/app/lib/pages/apps/app_detail/ARCHITECTURE.md b/app/lib/pages/apps/app_detail/ARCHITECTURE.md index 9aa136781a9..f2a47e77540 100644 --- a/app/lib/pages/apps/app_detail/ARCHITECTURE.md +++ b/app/lib/pages/apps/app_detail/ARCHITECTURE.md @@ -12,6 +12,12 @@ This package owns the mobile marketplace app-detail flow. Keep backend mutations in the owning page or section and pass refreshed state through callbacks. Do not duplicate install, subscription, or review mutation flows in widgets. Prefer a new focused file when a distinct section would otherwise make `app_detail.dart` grow. +External home and setup-step launches read fresh app details immediately before +opening. Setup grants are single-use; cached catalog/detail URLs are not a +fallback when refresh fails. The page and signed-in account still own the +response before adoption or launch, and the shared query-aware UID helper keeps +provider signatures intact. + ## Verification Run the app-detail and review widget tests, then `bash test.sh`. Exercise `app/e2e/flows/app-detail.yaml` on iOS and verify opening an app, returning to the marketplace, and the disabled/error path. diff --git a/app/lib/pages/apps/app_detail/app_detail.dart b/app/lib/pages/apps/app_detail/app_detail.dart index 4e03c555691..8949ec763be 100644 --- a/app/lib/pages/apps/app_detail/app_detail.dart +++ b/app/lib/pages/apps/app_detail/app_detail.dart @@ -48,6 +48,8 @@ class AppDetailPage extends StatefulWidget { final Future<(bool, String)> Function(String) enableApp; final Future?> Function(String) appDetailsLoader; final Future Function(String?) setupChecker; + final void Function(BuildContext context, App app) openAppHome; + final Future Function(Uri)? launchSetupUrl; const AppDetailPage({ super.key, @@ -56,8 +58,12 @@ class AppDetailPage extends StatefulWidget { this.enableApp = enableAppServer, this.appDetailsLoader = getAppDetailsServer, this.setupChecker = isAppSetupCompleted, + this.openAppHome = _routeToAppHome, + this.launchSetupUrl, }); + static void _routeToAppHome(BuildContext context, App app) => routeToPage(context, AppHomeWebPage(app: app)); + @override State createState() => _AppDetailPageState(); } @@ -95,13 +101,13 @@ class _AppDetailPageState extends State { } return true; } catch (e) { - Logger.warning('Failed to launch URL with in-app browser: $e'); + Logger.warning('Failed to launch app browser: ${e.runtimeType}'); // Fall back to external browser try { await launchUrl(uri, mode: LaunchMode.externalApplication); return true; } catch (e) { - Logger.warning('Failed to launch URL with external browser: $e'); + Logger.warning('Failed to launch external app browser: ${e.runtimeType}'); if (mounted) { OmiFeedback.error(context, context.l10n.couldNotOpenUrl); } @@ -117,9 +123,11 @@ class _AppDetailPageState extends State { } // TODO: move check to backend final generation = ++_setupCheckGeneration; + final uid = SharedPreferencesUtil().uid; final requestedUrl = app.externalIntegration!.setupCompletedUrl; widget.setupChecker(requestedUrl).then((value) { if (!mounted) return; + if (SharedPreferencesUtil().uid != uid) return; if (generation != _setupCheckGeneration) return; if (app.externalIntegration?.setupCompletedUrl != requestedUrl) return; @@ -224,12 +232,16 @@ class _AppDetailPageState extends State { ); WidgetsBinding.instance.addPostFrameCallback((_) async { - // Automatically open app home page if conditions are met - if (!widget.preventAutoOpenHomePage && app.enabled && app.externalIntegration?.appHomeUrl?.isNotEmpty == true) { - routeToPage(context, AppHomeWebPage(app: app)); + // Load details first: a transport-backed app's home URL carries a one-use + // launch grant that only a fresh details response contains. + final refreshed = await _refreshAppDetails(); + if (!mounted) return; + if (refreshed && + !widget.preventAutoOpenHomePage && + app.enabled && + app.externalIntegration?.appHomeUrl?.isNotEmpty == true) { + widget.openAppHome(context, app); } - // Load details - await _refreshAppDetails(); }); if (app.worksExternally()) { checkSetupCompleted(); @@ -239,9 +251,13 @@ class _AppDetailPageState extends State { super.initState(); } - Future _refreshAppDetails() async { + /// Returns whether fresh details were applied. + Future _refreshAppDetails() async { + final uid = SharedPreferencesUtil().uid; + final appId = app.id; setIsLoading(true); - var res = await context.read().getAppDetails(app.id); + var res = await context.read().getAppDetails(appId); + if (!mounted || SharedPreferencesUtil().uid != uid || app.id != appId) return false; if (mounted) { setState(() { if (res != null) { @@ -258,6 +274,44 @@ class _AppDetailPageState extends State { _loadSubscriptionData(); } } + return res != null; + } + + /// Opens the app's home page with a launch grant fetched for this open. A grant + /// is one-use and short-lived, so a catalog copy or an earlier response is never + /// reused. A failed refresh must not fall back to a consumed grant. + Future _openAppHome() async { + if (!await _refreshLaunchDetails() || !mounted) return; + if (app.externalIntegration?.appHomeUrl?.isNotEmpty != true) return; + widget.openAppHome(context, app); + } + + Future _refreshLaunchDetails() async { + final uid = SharedPreferencesUtil().uid; + final appId = app.id; + final fresh = await context.read().getAppDetails(appId); + if (!mounted || SharedPreferencesUtil().uid != uid || app.id != appId) return false; + if (fresh == null) { + OmiFeedback.error(context, context.l10n.unableToFetchApps); + return false; + } + setState(() => app = fresh); + return true; + } + + Future _openSetupStep(int index) async { + if (!await _refreshLaunchDetails() || !mounted) return; + final steps = app.externalIntegration?.authSteps ?? const []; + if (index < 0 || index >= steps.length) return; + final uri = Uri.tryParse(appSetupUrlWithUid(steps[index].url, SharedPreferencesUtil().uid)); + if (uri == null) { + OmiFeedback.error(context, context.l10n.invalidIntegrationUrl); + return; + } + final uid = SharedPreferencesUtil().uid; + if (await (widget.launchSetupUrl ?? _launchUrlSafely)(uri) && mounted && SharedPreferencesUtil().uid == uid) { + checkSetupCompleted(autoInstallIfCompleted: true); + } } void _onExternalIntegrationUpdated() { @@ -401,7 +455,7 @@ class _AppDetailPageState extends State { OmiIconButton.filled( icon: const FaIcon(FontAwesomeIcons.gear, size: 16), label: l10n.appSettingsLabel(app.name.decodeString), - onPressed: () => routeToPage(context, AppHomeWebPage(app: app)), + onPressed: _openAppHome, ), if (!isLoading && !app.private) Builder( @@ -497,15 +551,7 @@ class _AppDetailPageState extends State { AppSetupSteps( steps: app.externalIntegration!.authSteps, completed: setupCompleted, - onStepTap: (step) async { - final uri = Uri.tryParse(appSetupUrlWithUid(step.url, SharedPreferencesUtil().uid)); - if (uri == null) { - OmiFeedback.error(context, l10n.invalidIntegrationUrl); - return; - } - await _launchUrlSafely(uri); - checkSetupCompleted(autoInstallIfCompleted: true); - }, + onStepTap: (step) => _openSetupStep(app.externalIntegration!.authSteps.indexOf(step)), ), if (!hasAuthSteps && hasSetupInstructions) ListTile( @@ -726,6 +772,8 @@ class _AppDetailPageState extends State { } Future _navigateToSetup() async { + if (!await _refreshLaunchDetails() || !mounted) return; + final uid = SharedPreferencesUtil().uid; final authSteps = app.externalIntegration?.authSteps ?? const []; if (app.worksExternally() && authSteps.isNotEmpty) { final uri = Uri.tryParse(appSetupUrlWithUid(authSteps.first.url, SharedPreferencesUtil().uid)); @@ -733,11 +781,11 @@ class _AppDetailPageState extends State { if (mounted) OmiFeedback.error(context, context.l10n.invalidIntegrationUrl); return; } - await _launchUrlSafely(uri); + if (!await (widget.launchSetupUrl ?? _launchUrlSafely)(uri)) return; } else { await _openSetupInstructions(); } - _startSetupCompletionCheck(); + if (mounted && SharedPreferencesUtil().uid == uid) _startSetupCompletionCheck(); } /// Opens an integration's setup instructions: rendered markdown, an external link, or inline text. @@ -838,9 +886,7 @@ class _AppDetailPageState extends State { // Automatically open app home page after installation if available if (app.externalIntegration?.appHomeUrl?.isNotEmpty == true) { Future.delayed(const Duration(seconds: 1), () { - if (mounted) { - routeToPage(context, AppHomeWebPage(app: app)); - } + if (mounted) _openAppHome(); }); } } diff --git a/app/lib/providers/app_provider.dart b/app/lib/providers/app_provider.dart index 37fc23bdf03..d5e7a0f4047 100644 --- a/app/lib/providers/app_provider.dart +++ b/app/lib/providers/app_provider.dart @@ -33,6 +33,9 @@ class AppProvider extends BaseProvider { @visibleForTesting AppsSearchRequest? searchAppsOverride; + @visibleForTesting + Future?> Function(String)? appDetailsOverride; + /// Test seam — overrides [enableAppServer] in [toggleApp]. @visibleForTesting Future<(bool, String)> Function(String appId)? enableAppOverride; @@ -77,6 +80,7 @@ class AppProvider extends BaseProvider { Future? _appsLoad; Future? _popularAppsLoad; int _activeCatalogLoads = 0; + int _detailSession = 0; List categories = []; List capabilities = []; @@ -105,18 +109,18 @@ class AppProvider extends BaseProvider { } Future getAppDetails(String id) async { - var appData = await getAppDetailsServer(id); - if (appData != null) { - var freshApp = App.fromJson(appData); - var oldApp = apps.where((element) => element.id == id).firstOrNull; - if (oldApp != null) { - var idx = apps.indexOf(oldApp); - apps[idx] = freshApp; - notifyListeners(); - } - return freshApp; + final session = _detailSession; + final uid = SharedPreferencesUtil().uid; + final data = await (appDetailsOverride ?? getAppDetailsServer)(id); + if (data == null || session != _detailSession || uid != SharedPreferencesUtil().uid) return null; + final fresh = App.fromJson(data); + if (fresh.id != id) return null; + final index = apps.indexWhere((app) => app.id == id); + if (index >= 0) { + apps[index] = fresh; + notifyListeners(); } - return null; + return fresh; } void updateInstalledAppsOptionSelected(bool value) { @@ -220,6 +224,7 @@ class AppProvider extends BaseProvider { } void clearUserData() { + _detailSession++; apps = []; popularApps = []; groupedApps = []; diff --git a/app/test/providers/plugin_launch_session_test.dart b/app/test/providers/plugin_launch_session_test.dart new file mode 100644 index 00000000000..8b785cb03ce --- /dev/null +++ b/app/test/providers/plugin_launch_session_test.dart @@ -0,0 +1,31 @@ +import 'dart:async'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:shared_preferences/shared_preferences.dart'; +import 'package:omi/backend/preferences.dart'; +import 'package:omi/providers/app_provider.dart'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + setUp(() async { + SharedPreferences.setMockInitialValues({}); + await SharedPreferencesUtil.init(); + }); + + for (final replacement in ['other-account', 'original-account']) { + test('retired launch details cannot populate a replacement/ABA catalog: $replacement', () async { + SharedPreferencesUtil().uid = 'original-account'; + final pending = Completer?>(); + final provider = AppProvider()..appDetailsOverride = (_) => pending.future; + addTearDown(provider.dispose); + final read = provider.getAppDetails('app-1'); + provider.clearUserData(); + SharedPreferencesUtil().uid = replacement; + pending.complete({ + 'id': 'app-1', + 'external_integration': {'app_home_url': 'https://plugin.test/old-grant'} + }); + expect(await read, isNull); + expect(provider.apps, isEmpty); + }); + } +} diff --git a/app/test/unit/backend/http/api_secret_log_safety_test.dart b/app/test/unit/backend/http/api_secret_log_safety_test.dart index d91a772f862..243d27e834a 100644 --- a/app/test/unit/backend/http/api_secret_log_safety_test.dart +++ b/app/test/unit/backend/http/api_secret_log_safety_test.dart @@ -32,6 +32,17 @@ final _responses = { ], '/v1/apps/mcp:POST': {'app_id': 'mcp-1', 'requires_oauth': true, 'auth_url': 'https://idp.invalid/a?state=$_secret'}, '/v1/phone/token:POST': {'access_token': _secret, 'identity': 'user-1', 'ttl': 3600}, + '/v1/apps/app-1:GET': { + 'id': 'app-1', + 'name': 'Backup', + 'author': 'Omi', + 'description': 'Backup fixture', + 'category': 'productivity', + 'image': '', + 'capabilities': ['external_integration'], + 'external_integration': {'auth_steps': [], 'app_home_url': 'https://plugin.invalid/setup?omi_token=$_secret'}, + }, + '/plugin/status:GET': {'is_setup_completed': true}, }; /// Runs [body] while capturing every channel a log line can leave through: the app's [Logger] @@ -108,4 +119,33 @@ void main() { final logs = await _captureLogs(() async => expect((await getPhoneCallToken()).token, isNotNull)); expect(logs, isNot(contains(_secret))); }); + + test('fresh app details return the launch grant without logging it', () async { + Map? detail; + final logs = await _captureLogs(() async => detail = await getAppDetailsServer('app-1')); + expect((detail?['external_integration'] as Map)['app_home_url'], contains(_secret)); + expect(logs, contains('getAppDetailsServer: 200')); + expect(logs, isNot(contains(_secret))); + }); + + test('vendor completion uses the real HTTP pool without forwarding an Omi bearer or logging the grant', () async { + final vendor = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + String? authorization; + vendor.listen((request) async { + authorization = request.headers.value(HttpHeaders.authorizationHeader); + expect(request.uri.queryParameters['omi_token'], _secret); + request.response.headers.contentType = ContentType.json; + request.response.write('{"is_setup_completed":true}'); + await request.response.close(); + }); + try { + final logs = await _captureLogs(() async => expect( + await isAppSetupCompleted('http://${vendor.address.host}:${vendor.port}/plugin/status?omi_token=$_secret'), + isTrue)); + expect(authorization, isNull); + expect(logs, isNot(contains(_secret))); + } finally { + await vendor.close(force: true); + } + }); } diff --git a/app/test/unit/backend/plugin_setup_authorization_test.dart b/app/test/unit/backend/plugin_setup_authorization_test.dart new file mode 100644 index 00000000000..0d2d215356f --- /dev/null +++ b/app/test/unit/backend/plugin_setup_authorization_test.dart @@ -0,0 +1,44 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:http/http.dart' as http; +import 'package:omi/backend/http/api/apps.dart'; + +void main() { + test('plugin completion uses the signed query without Omi bearer headers', () async { + final result = await isAppSetupCompleted( + 'https://plugin.test/status?omi_token=synthetic.signed&uid=old', + uid: 'fixture-user', + send: (request) async { + expect(Uri.parse(request.url).queryParameters, {'omi_token': 'synthetic.signed', 'uid': 'fixture-user'}); + expect(request.headers, isEmpty); + return http.Response('{"is_setup_completed":true}', 200); + }, + ); + expect(result, isTrue); + }); + + test('a vendor rejection or malformed verdict never completes setup', () async { + for (final (status, body) in [ + (401, '{"is_setup_completed":true}'), + (503, '{"is_setup_completed":true}'), + (200, '{"is_setup_completed":"true"}'), + (200, '{}'), + (200, 'not json'), + ]) { + expect( + await isAppSetupCompleted('https://plugin.test/status', + uid: 'fixture-user', send: (_) async => http.Response(body, status)), + isFalse); + } + }); + + test('legacy apps without a completion endpoint retain optional setup', () async { + var sends = 0; + expect( + await isAppSetupCompleted(null, uid: 'fixture-user', send: (_) async { + sends++; + return http.Response('{}', 200); + }), + isTrue); + expect(sends, 0); + }); +} diff --git a/app/test/widgets/app_install_flow_test.dart b/app/test/widgets/app_install_flow_test.dart index 0bbab24bab6..7a7682d9838 100644 --- a/app/test/widgets/app_install_flow_test.dart +++ b/app/test/widgets/app_install_flow_test.dart @@ -19,9 +19,10 @@ class _FakeAppProvider extends AppProvider { _FakeAppProvider(this.detail); final App detail; + Future Function()? load; @override - Future getAppDetails(String id) async => detail; + Future getAppDetails(String id) async => load == null ? detail : await load!(); } App _app({ @@ -107,11 +108,13 @@ void main() { Future<(bool, String)> Function(String)? enableApp, Future?> Function(String)? appDetailsLoader, Future Function(String?)? setupChecker, + AppProvider? appProvider, + Future Function(Uri)? launchSetupUrl, }) async { await tester.pumpWidget( MultiProvider( providers: [ - ChangeNotifierProvider.value(value: _FakeAppProvider(app)), + ChangeNotifierProvider.value(value: appProvider ?? _FakeAppProvider(app)), ChangeNotifierProvider.value(value: AddAppProvider()), ChangeNotifierProvider.value(value: MessageProvider()), ], @@ -124,6 +127,7 @@ void main() { enableApp: enableApp ?? (_) async => (false, ''), appDetailsLoader: appDetailsLoader ?? (_) async => null, setupChecker: setupChecker ?? (_) async => false, + launchSetupUrl: launchSetupUrl, ), ), ), @@ -375,4 +379,138 @@ void main() { expect(enableCalls, 0, reason: 'missing payment link must not attempt an unpaid install'); expect(find.text('Invalid payment URL'), findsOneWidget); }); + + group('app home with a one-use launch grant', () { + App homeApp(String homeUrl, {bool enabled = true}) => App( + id: 'app-1', + name: 'Backups', + author: 'Omi', + description: 'Backs up conversations', + image: '', + capabilities: {'external_integration'}, + status: 'approved', + category: 'productivity', + approved: true, + ratingCount: 0, + enabled: enabled, + deleted: false, + isPaid: false, + isUserPaid: false, + externalIntegration: ExternalIntegration(appHomeUrl: homeUrl), + ); + + Future> pumpHome(WidgetTester tester, App catalog, App fresh, + {bool preventAutoOpen = false, Future<(bool, String)> Function(String)? enableApp}) async { + final opened = []; + await tester.pumpWidget(MultiProvider( + providers: [ + ChangeNotifierProvider.value(value: _FakeAppProvider(fresh)), + ChangeNotifierProvider.value(value: AddAppProvider()), + ChangeNotifierProvider.value(value: MessageProvider()), + ], + child: MaterialApp( + localizationsDelegates: AppLocalizations.localizationsDelegates, + supportedLocales: const [Locale('en')], + home: AppDetailPage( + app: catalog, + preventAutoOpenHomePage: preventAutoOpen, + enableApp: enableApp ?? (_) async => (true, ''), + appDetailsLoader: (_) async => null, + setupChecker: (_) async => false, + openAppHome: (_, app) => opened.add(app.externalIntegration!.appHomeUrl!), + ), + ), + )); + await settle(tester); + return opened; + } + + testWidgets('auto-open uses the freshly fetched grant, never the catalog URL', (tester) async { + final opened = await pumpHome( + tester, homeApp('https://plugin.test/home'), homeApp('https://plugin.test/home?omi_token=fresh-1')); + expect(opened, ['https://plugin.test/home?omi_token=fresh-1']); + }); + + testWidgets('the settings button fetches a new grant for each open', (tester) async { + final opened = await pumpHome(tester, homeApp('https://plugin.test/home?omi_token=cached'), + homeApp('https://plugin.test/home?omi_token=fresh-2'), + preventAutoOpen: true); + expect(opened, isEmpty); + await tester.tap(find.byTooltip('Backups settings')); + await settle(tester); + expect(opened, ['https://plugin.test/home?omi_token=fresh-2']); + }); + + testWidgets('opening after install uses a fresh grant', (tester) async { + final opened = await pumpHome(tester, homeApp('https://plugin.test/home', enabled: false), + homeApp('https://plugin.test/home?omi_token=fresh-3', enabled: false), + preventAutoOpen: true); + await tester.tap(find.text('Enable')); + await settle(tester); + await tester.tap(find.text('Enable').last); // the data-access consent dialog + await tester.pump(const Duration(seconds: 2)); + await settle(tester); + expect(opened, ['https://plugin.test/home?omi_token=fresh-3']); + }); + }); + + group('fresh setup capabilities', () { + App setupApp(int grant) => _app( + external: true, + authSteps: [ + AuthStep(name: 'Connect pCloud', url: 'https://plugin.test/setup?uid=u1&omi_token=signed.$grant') + ], + ); + + testWidgets('each setup-step click fetches a fresh grant with intact query', (tester) async { + var reads = 0; + final provider = _FakeAppProvider(setupApp(0))..load = () async => setupApp(++reads); + final opened = []; + await pumpDetail(tester, setupApp(0), appProvider: provider, launchSetupUrl: (uri) async { + opened.add(uri); + return true; + }); + await settle(tester); + await tester.ensureVisible(find.text('Connect pCloud')); + await tester.tap(find.text('Connect pCloud')); + await settle(tester); + await tester.tap(find.text('Connect pCloud')); + await settle(tester); + expect(opened.map((uri) => uri.queryParameters['omi_token']), ['signed.2', 'signed.3']); + expect(opened.every((uri) => uri.queryParametersAll['uid']!.length == 1), isTrue); + }); + + testWidgets('failed refresh never opens the cached setup grant', (tester) async { + final provider = _FakeAppProvider(setupApp(0)); + var launches = 0; + await pumpDetail(tester, setupApp(0), appProvider: provider, launchSetupUrl: (_) async { + launches++; + return true; + }); + await settle(tester); + provider.load = () async => null; + await tester.ensureVisible(find.text('Connect pCloud')); + await tester.tap(find.text('Connect pCloud')); + await settle(tester); + expect(launches, 0); + }); + + testWidgets('account replacement while refreshing cannot open the previous grant', (tester) async { + final provider = _FakeAppProvider(setupApp(0)); + var launches = 0; + await pumpDetail(tester, setupApp(0), appProvider: provider, launchSetupUrl: (_) async { + launches++; + return true; + }); + await settle(tester); + final pending = Completer(); + provider.load = () => pending.future; + await tester.ensureVisible(find.text('Connect pCloud')); + await tester.tap(find.text('Connect pCloud')); + SharedPreferencesUtil().uid = 'another-account'; + pending.complete(setupApp(1)); + await settle(tester); + expect(launches, 0); + }); + }); } diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 6847f02a77e..249fa4756ba 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -95,6 +95,8 @@ backend/ ## Service Map +[Plugin transport](docs/plugin-transport.md). + ``` Shared: Firestore, Redis diff --git a/backend/docs/listen_pusher_pipeline.mdx b/backend/docs/listen_pusher_pipeline.mdx index 6142de75ab1..bb96fa1d076 100644 --- a/backend/docs/listen_pusher_pipeline.mdx +++ b/backend/docs/listen_pusher_pipeline.mdx @@ -589,6 +589,16 @@ retains the pending item for reconnect replay; buffers are cleared only after th send completes. Finalization opcode `104` carries the durable job id and dispatch generation, and Pusher claims that Firestore-owned job before processing. +The opt-in pCloud backup completion (#20213/#20498) uses finalized conversation +fanout, not the live UID-only raw-audio webhook. Registered transports sign the +account, destination path and redacted payload, preserving existing idempotency +and SSRF pinning. The standalone service acknowledges only durable admission, +then exports consented files through leased, per-file-checkpointed jobs. Audio +reads use canonical saved-audio identity and recheck account/app/conversation +access; no independent buffer may guess conversation boundaries or retain a +discarded recording for the next conversation. Architecture, unprovisioned +runtime bindings and release checklist: [`plugin-transport.md`](plugin-transport.md). + After a finalization lease is acquired, `finalizer.py` may admit a Parakeet **shadow** pass before canonical processing. Admission schedules a tracked task on the finalizer host: the pusher GKE pod for live WebSocket handoff, or diff --git a/backend/docs/plugin-transport.md b/backend/docs/plugin-transport.md new file mode 100644 index 00000000000..64224c095c4 --- /dev/null +++ b/backend/docs/plugin-transport.md @@ -0,0 +1,98 @@ +# Authenticated standalone-plugin transport + +Owner: `utils/plugin_transport.py`. First consumer: the pCloud completion of +#20498; delivery problem and acceptance are tracked in #20213. + +## Authority and composition + +Authenticated `GET /v1/apps/{app_id}` decorates a deep copy of the app's setup, +home and completion URLs with per-account capabilities. Cached/public marketplace +records never receive them; owner edits strip launch credentials before storage. +The existing Flutter/macOS setup flows already use these fields and append the +same UID hint, so no new client-specific auth channel or mobile-only state store +is introduced. The plugin consumes setup nonces atomically and uses an HttpOnly +session plus CSRF-protected POST forms for mutations. A query UID never authorizes. +A launch grant is one-use and short-lived, so every client fetches fresh details +immediately before each home-page or authentication-step open instead of reusing a catalog or earlier +response: the Flutter app detail page (auto-open, the settings button and +post-install), the macOS and Windows app sheets, the Windows Apps page's first-time +Install, and the web app detail page. +Refresh failure never falls back to an earlier capability. Query-aware UID +helpers preserve signed query bytes, and late reads for a departed account or +page cannot launch the browser. Setup retries use the existing client owners, +not another authorization endpoint or parallel token cache. + +Finalized conversation fanout preserves SSRF target pinning, privacy redaction, +the existing durable idempotency key and legacy apps. Only a registered app gets +a signed delivery header and conversation-scoped saved-audio capability. Its +existing webhook URL must match the trusted registration origin; metadata edits +cannot redirect a capability to a different host. Re-enabling an auto-disabled +registered app checks that origin and probes the service's non-mutating +`GET /health`, because the authenticated delivery endpoint rightly rejects the +unsigned empty `POST` legacy webhooks receive; deliveries stay signed. + +The plugin durably admits a job before acknowledging delivery. Its persistent +store owns idempotency, per-file receipts, retry state, consent and connection +generation. It is not a new Omi conversation or audio source of truth. It reads +canonical saved audio through the existing playback path instead of attempting +to join uncorrelated live raw-audio callbacks. Missing audio is explicit. + +`HEAD /v1/apps/{app_id}/backup-audio/{conversation_id}` rechecks current account +deletion, enabled-app availability and retained/unlocked/non-discarded conversation +ownership before a pending provider write. It does not inspect or enqueue audio. +`GET` applies the same checks and delegates the saved-audio manifest to +`utils/sync/playback.py`; it never accepts a caller-provided UID or download URL. +These are internal plugin-protocol endpoints, excluded from released-client SDKs. + +## Wire protocol + +Per-app HS256 JWTs use issuer `omi-plugin-transport/v1`, audience = app ID, +`sub` = authenticated UID, `iat`, `exp`, random `jti` and an exact `scope`: + +| Scope | Lifetime | Binding | +| --- | --- | --- | +| `setup` | 10 minutes, consumed once | Registered HTTPS service and account | +| `status` | 10 minutes, repeatable read | Connected-state boolean for that account | +| `delivery` | 10 minutes | Account, POST path, canonical JSON SHA-256 | +| `audio` | 7 days maximum | Account and one conversation; fresh access checked on HEAD/GET | + +Setup/status URL parameter: `omi_token`. Delivery header: +`X-Omi-Plugin-Authorization` (raw JWT). Canonical JSON is UTF-8 encoding of +`json.dumps(payload, sort_keys=True, separators=(',', ':'), ensure_ascii=True, +allow_nan=False)`. The signed payload includes `_omi_backup_audio_token`. +Audio/access reads use `Authorization: Bearer `. + +The pCloud verifier is independently deployed and uses the same explicit wire +contract, tested against the actual backend signer. The models-only plugin SDK +is not expanded into an auth/worker framework. No unsigned fallback is accepted +by this new service. Old unregistered apps retain their existing behavior; this +does not retroactively secure all other plugins. + +## Configuration and rollout boundary + +`OMI_PLUGIN_TRANSPORT_JSON` is a **server-only secret**, mapping registered app IDs +to `base_url` and a distinct random `key` of at least 32 bytes. No value is +provisioned here. The pCloud service gets only its own app ID and key. Existing +backend runtime releases do not acquire a new required secret from this PR. +Before enabling a registration, its optional runtime binding must be reviewed +and provisioned across every host that executes finalized conversation fanout. +That deployment/configuration work is intentionally not performed by this +source-only task. The completion PR remains draft until the release checklist +is satisfied; do not merge an unsigned foundation first as a workaround. + +Do not log capability-bearing URLs, header values, provider error bodies, +credentials or private payloads. The standalone app/proxy disable query access +logs. Signing-key rotation requires coordinated admission pause and handling of +outstanding jobs; encryption-key rotation is separate and tested in the plugin. + +Architecture/implementation checklist and durable-storage topology: +[`plugins/omi-pcloud-app/README.md`](../../plugins/omi-pcloud-app/README.md). + +## Verification ownership + +The manifest-backed pCloud suite exercises real HTTP routing, cookie/CSRF/expiry, +body/account binding, durable partial success, consent, revocation and key rotation. +Backend tests exercise metadata copying, registered-origin restrictions, actual +fanout composition, legacy behavior and current access checks. These are +hermetic; live pCloud grant/callback and volume recovery verification belong to +the deploying operator. No live verification is implied by mocked provider IO. diff --git a/backend/docs/route_policy_manifest.md b/backend/docs/route_policy_manifest.md index 4a0395321b4..faaade19589 100644 --- a/backend/docs/route_policy_manifest.md +++ b/backend/docs/route_policy_manifest.md @@ -83,3 +83,5 @@ On pull requests, CI compares this file with the target branch copy and fails if 6. Use `review_status: exempt` only with an `exempt_reason`. Keep declared policy separate from observed evidence. Dependency names, OpenAPI tags, timeout override hints, and endpoint modules are generated to help review, but they are not a substitute for owner-reviewed policy. + +`plugin_capability` denotes the opt-in plugin transport's app/account/conversation-scoped HS256 bearer, not a Firebase or Google OIDC token. The audio handler also checks current account, installation and conversation eligibility; see [plugin transport](plugin-transport.md). diff --git a/backend/route_policy_manifest.yaml b/backend/route_policy_manifest.yaml index 184c17f05e3..699448dca45 100644 --- a/backend/route_policy_manifest.yaml +++ b/backend/route_policy_manifest.yaml @@ -6,6 +6,10 @@ schema_version: 1 service: backend-main routes: + # Exact app/account/conversation-scoped HS256 capability, followed by current + # account, installation and conversation eligibility checks in the handler. + - {route_type: http, method: GET, path: "/v1/apps/{app_id}/backup-audio/{conversation_id}", policy: {review_status: reviewed, auth: {mechanisms: [plugin_capability], placement: handler, scopes: [audio]}, byok: not_applicable, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: default_method, surface: developer_api, visibility: internal, data_domain: sync_audio, deprecation: {state: active}, owner: backend}} + - {route_type: http, method: HEAD, path: "/v1/apps/{app_id}/backup-audio/{conversation_id}", policy: {review_status: reviewed, auth: {mechanisms: [plugin_capability], placement: handler, scopes: [audio]}, byok: not_applicable, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: default_method, surface: developer_api, visibility: internal, data_domain: sync_audio, deprecation: {state: active}, owner: backend}} - {route_type: http, method: POST, path: /v1/capture/boundary-evidence, policy: {review_status: reviewed, auth: {mechanisms: [firebase_id_token, admin_key_uid_prefix], placement: dependency, scopes: []}, byok: not_applicable, rate_limit: {policy_name: "episodes:write", key_subject: uid, enforcement: fail_open, placement: dependency}, timeout_class: default_method, surface: first_party_app, visibility: first_party, data_domain: conversations, deprecation: {state: active}, owner: backend}} - {route_type: http, method: GET, path: /v2/capture-coverage, policy: {review_status: reviewed, auth: {mechanisms: [firebase_id_token, admin_key_uid_prefix], placement: dependency, scopes: []}, byok: not_applicable, rate_limit: {policy_name: "capture_coverage:query", key_subject: uid, enforcement: fail_closed, placement: inline}, timeout_class: default_method, surface: first_party_app, visibility: first_party, data_domain: conversations, deprecation: {state: active}, owner: backend}} - {route_type: http, method: POST, path: /v2/cap-hold-recovery/sweep, policy: {review_status: reviewed, auth: {mechanisms: [service_oidc], placement: dependency, scopes: []}, byok: not_applicable, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: sync_job, surface: internal_task, visibility: internal, data_domain: sync_audio, deprecation: {state: active}, owner: backend}} diff --git a/backend/routers/apps.py b/backend/routers/apps.py index 5da7d5d30e8..0abc0bf01d4 100644 --- a/backend/routers/apps.py +++ b/backend/routers/apps.py @@ -31,6 +31,8 @@ safe_request_targets, ) from utils.multipart import APP_IMAGE_MAX_PART_SIZE, MultipartMaxPartSizeRoute, max_part_size +from utils.plugin_transport import authorize_app_setup, plugin_transport, strip_launch_credentials +from utils.plugin_backup_audio import backup_audio_manifest from utils.mcp_client import ( discover_oauth_metadata, register_oauth_client, @@ -1150,6 +1152,7 @@ def update_app( # Backfill app_home_url from auth steps; explicit null clears the integration, non-dicts fail validation. if isinstance(data.get('external_integration'), dict): + strip_launch_credentials(app_id, data['external_integration']) backfill_app_home_url_from_auth_steps(data['external_integration']) _set_instructions_url_flag(data['external_integration']) @@ -1292,7 +1295,19 @@ def get_app_details(app_id: str, uid: str = Depends(auth.get_current_user_uid)): if app.thumbnails: app.thumbnail_urls = [get_app_thumbnail_url(thumbnail_id) for thumbnail_id in app.thumbnails] - return app + try: + return authorize_app_setup(app, uid) + except ValueError: + raise HTTPException( + status_code=503, detail='This app connection is not configured correctly. Please try later.' + ) + + +@router.get('/v1/apps/{app_id}/backup-audio/{conversation_id}', include_in_schema=False) +@router.head('/v1/apps/{app_id}/backup-audio/{conversation_id}', include_in_schema=False) +def get_plugin_backup_audio(app_id: str, conversation_id: str, request: Request, authorization: str = Header('')): + """Internal, conversation-scoped plugin read; never accepts a query UID.""" + return backup_audio_manifest(app_id, conversation_id, authorization, include_audio=request.method == 'GET') @router.get('/v1/app-categories', tags=['v1'], response_model=List[AppSelectOption]) @@ -2322,6 +2337,14 @@ async def enable_app_endpoint(app_id: str, request: Request, uid: str = Depends( if app.external_integration.setup_completed_url: client = get_pinned_delivery_client() setup_url = app.external_integration.setup_completed_url + try: + transport = plugin_transport(app.id) + if transport: + setup_url = transport.setup_url(setup_url, uid, 'status') + except ValueError: + raise HTTPException( + status_code=503, detail='This app connection is not configured correctly. Please try later.' + ) # Developer-controlled URL: validate and pin it exactly like the OAuth setup check. # Resolve once, reject private/loopback/reserved targets, then connect to the resolved # IP while presenting the original Host/SNI, so a DNS record swapped between check and diff --git a/backend/scripts/route_policy_inventory.py b/backend/scripts/route_policy_inventory.py index 76576f2408a..7dfda643f71 100644 --- a/backend/scripts/route_policy_inventory.py +++ b/backend/scripts/route_policy_inventory.py @@ -60,6 +60,7 @@ 'service_oidc', 'google_compute_identity_token', 'webhook_signature', + 'plugin_capability', 'websocket_first_message', 'unknown', } diff --git a/backend/tests/unit/test_app_auto_disable_recovery.py b/backend/tests/unit/test_app_auto_disable_recovery.py index 467726898fa..f3d204b34f1 100644 --- a/backend/tests/unit/test_app_auto_disable_recovery.py +++ b/backend/tests/unit/test_app_auto_disable_recovery.py @@ -19,6 +19,8 @@ from unittest.mock import MagicMock, patch +import json + import httpx import pytest from fastapi import HTTPException @@ -175,6 +177,42 @@ def test_health_check_does_not_follow_redirects(self): 'followed passes endpoints that fail on the first real webhook' ) + TRANSPORT_APP = 'synthetic-pcloud' + TRANSPORT_BASE = 'https://backup.example.invalid' + + def _transport(self, monkeypatch): + monkeypatch.setenv( + 'OMI_PLUGIN_TRANSPORT_JSON', + json.dumps({self.TRANSPORT_APP: {'base_url': self.TRANSPORT_BASE, 'key': 'k' * 40}}), + ) + + def test_transport_app_is_probed_by_health_never_an_unsigned_delivery(self, monkeypatch): + self._transport(monkeypatch) + app = {'external_integration': {'webhook_url': self.TRANSPORT_BASE + '/conversation'}} + with patch('utils.apps.httpx.request', return_value=httpx.Response(200)) as request: + validate_app_endpoints_for_reenable(app, {}, self.TRANSPORT_APP) + + (method, url), kwargs = request.call_args + assert (method, url) == ('GET', self.TRANSPORT_BASE + '/health') + assert kwargs['follow_redirects'] is False + + def test_unhealthy_transport_app_stays_disabled(self, monkeypatch): + self._transport(monkeypatch) + app = {'external_integration': {'webhook_url': self.TRANSPORT_BASE + '/conversation'}} + with patch('utils.apps.httpx.request', return_value=httpx.Response(503)): + with pytest.raises(HTTPException) as exc: + validate_app_endpoints_for_reenable(app, {}, self.TRANSPORT_APP) + assert exc.value.status_code == 400 + + def test_transport_webhook_off_its_trusted_origin_is_rejected(self, monkeypatch): + self._transport(monkeypatch) + app = {'external_integration': {'webhook_url': 'https://elsewhere.example.invalid/conversation'}} + with patch('utils.apps.httpx.request', return_value=httpx.Response(200)) as request: + with pytest.raises(HTTPException) as exc: + validate_app_endpoints_for_reenable(app, {}, self.TRANSPORT_APP) + assert exc.value.status_code == 400 + request.assert_not_called() + def test_redirecting_webhook_is_rejected_with_an_actionable_reason(self): app = {'external_integration': {'webhook_url': 'https://dev.example.com/omi/webhook'}} with patch('utils.apps.httpx.request', return_value=httpx.Response(307)): diff --git a/backend/tests/unit/test_plugin_transport.py b/backend/tests/unit/test_plugin_transport.py new file mode 100644 index 00000000000..311df72dc8a --- /dev/null +++ b/backend/tests/unit/test_plugin_transport.py @@ -0,0 +1,172 @@ +"""Pure transport and real scoped-audio boundary tests, without cloud services.""" + +import copy +import json +from types import ModuleType, SimpleNamespace + +import jwt +import pytest +from fastapi import HTTPException + +from testing.import_isolation import load_module_fresh, stub_modules +from utils.plugin_transport import PluginTransport, authorize_app_setup, plugin_transport, strip_launch_credentials + +KEY = 'synthetic-transport-secret-with-32-bytes' +APP = 'synthetic-pcloud' +TRANSPORT = PluginTransport(APP, 'https://backup.example.invalid', KEY) + + +@pytest.mark.parametrize('raw', ['', '{}']) +def test_unregistered_legacy_app_is_unchanged(monkeypatch, raw): + monkeypatch.setenv('OMI_PLUGIN_TRANSPORT_JSON', raw) + app = SimpleNamespace(id='legacy', external_integration=None) + assert plugin_transport('legacy') is None + assert authorize_app_setup(app, 'user') is app + + +@pytest.mark.parametrize( + 'raw', + [ + '[]', + '{bad', + json.dumps({APP: {'base_url': 'http://localhost', 'key': KEY}}), + json.dumps({APP: {'base_url': 'https://backup.example.invalid', 'key': 'short'}}), + ], +) +def test_bad_registration_fails_closed(monkeypatch, raw): + monkeypatch.setenv('OMI_PLUGIN_TRANSPORT_JSON', raw) + with pytest.raises(ValueError): + plugin_transport(APP) + + +def test_setup_urls_are_per_account_deep_copies(monkeypatch): + monkeypatch.setenv('OMI_PLUGIN_TRANSPORT_JSON', json.dumps({APP: {'base_url': TRANSPORT.base_url, 'key': KEY}})) + + class App: + id = APP + external_integration = SimpleNamespace( + auth_steps=[SimpleNamespace(url=TRANSPORT.base_url + '/setup/pcloud')], + app_home_url=TRANSPORT.base_url + '/setup/pcloud', + setup_completed_url=TRANSPORT.base_url + '/setup-completed', + ) + + def model_copy(self, deep): + return copy.deepcopy(self) + + app = App() + app.external_integration = copy.deepcopy(App.external_integration) + a = authorize_app_setup(app, 'user-a') + b = authorize_app_setup(app, 'user-b') + assert 'omi_token=' not in app.external_integration.auth_steps[0].url + assert a.external_integration.auth_steps[0].url != b.external_integration.auth_steps[0].url + assert 'uid=user-a' in a.external_integration.app_home_url + assert 'uid=user-b' in b.external_integration.setup_completed_url + + +@pytest.mark.parametrize( + 'url', + [ + 'https://evil.invalid/setup', + 'http://backup.example.invalid/setup', + 'https://backup.example.invalid.evil.invalid/setup', + 'https://user:secret@backup.example.invalid/setup', + 'https://backup.example.invalid/setup#secret', + ], +) +def test_signed_setup_and_delivery_never_follow_untrusted_metadata(url): + with pytest.raises(ValueError): + TRANSPORT.setup_url(url, 'user') + with pytest.raises(ValueError): + TRANSPORT.delivery_headers('user', url, {}) + + +def test_delivery_binds_body_path_and_principal(): + headers = TRANSPORT.delivery_headers( + 'user-a', TRANSPORT.base_url + '/conversation?uid=user-a', {'id': 'conversation'} + ) + claims = TRANSPORT.decode(headers['X-Omi-Plugin-Authorization'], 'delivery') + assert claims['sub'] == 'user-a' and claims['path'] == '/conversation' + assert len(claims['sha256']) == 64 + with pytest.raises(ValueError): + TRANSPORT.decode(headers['X-Omi-Plugin-Authorization'], 'setup') + + +def test_owner_edit_cannot_persist_a_launch_capability(monkeypatch): + monkeypatch.setenv('OMI_PLUGIN_TRANSPORT_JSON', json.dumps({APP: {'base_url': TRANSPORT.base_url, 'key': KEY}})) + metadata = { + 'auth_steps': [{'url': TRANSPORT.setup_url(TRANSPORT.base_url + '/setup/pcloud?theme=light', 'user')}], + 'app_home_url': TRANSPORT.setup_url(TRANSPORT.base_url + '/setup/pcloud', 'user'), + } + strip_launch_credentials(APP, metadata) + assert metadata['auth_steps'][0]['url'] == TRANSPORT.base_url + '/setup/pcloud?theme=light' + assert metadata['app_home_url'] == TRANSPORT.base_url + '/setup/pcloud' + + +@pytest.fixture +def audio_boundary(): + calls = [] + + def module(name, **attrs): + value = ModuleType(name) + value.__dict__.update(attrs) + return value + + dependencies = { + 'database.conversations': module( + 'database.conversations', + get_conversation=lambda uid, cid: {'id': cid, 'audio_files': [{'id': 'audio-part'}]}, + ), + 'utils.apps': module( + 'utils.apps', + is_user_app_enabled=lambda uid, app: True, + get_available_app_by_id=lambda app, uid: {'id': app}, + ), + 'utils.other.endpoints': module( + 'utils.other.endpoints', enforce_account_deletion_http_access=lambda uid: calls.append(('deletion', uid)) + ), + 'utils.sync.playback': module( + 'utils.sync.playback', + get_audio_signed_urls=lambda uid, cid, files, conversation: calls.append(('audio', uid, cid)) + or {'audio_files': []}, + ), + } + with stub_modules(dependencies): + boundary = load_module_fresh('utils.plugin_backup_audio', 'utils/plugin_backup_audio.py') + boundary.plugin_transport = lambda app: TRANSPORT if app == APP else None + yield boundary, calls + + +def test_audio_read_is_owner_and_conversation_scoped(audio_boundary): + boundary, calls = audio_boundary + token = TRANSPORT.token('user-a', 'audio', conversation_id='conversation-a') + assert boundary.backup_audio_manifest(APP, 'conversation-a', 'Bearer ' + token) == {'audio_files': []} + assert calls == [('deletion', 'user-a'), ('audio', 'user-a', 'conversation-a')] + with pytest.raises(HTTPException) as exc: + boundary.backup_audio_manifest(APP, 'conversation-b', 'Bearer ' + token) + assert exc.value.status_code == 401 + + +def test_head_access_probe_does_not_trigger_audio_work(audio_boundary): + boundary, calls = audio_boundary + token = TRANSPORT.token('user-a', 'audio', conversation_id='conversation-a') + assert boundary.backup_audio_manifest(APP, 'conversation-a', 'Bearer ' + token, include_audio=False) == { + 'authorized': True + } + assert calls == [('deletion', 'user-a')] + + +@pytest.mark.parametrize('status', ['disabled', 'discarded', 'locked', 'deleted', 'missing']) +def test_audio_rechecks_current_access_not_only_grant(audio_boundary, status): + boundary, calls = audio_boundary + if status == 'disabled': + boundary.is_user_app_enabled = lambda *args: False + elif status == 'missing': + boundary.get_conversation = lambda *args: None + else: + key = 'is_locked' if status == 'locked' else status + boundary.get_conversation = lambda *args: {key: True} + token = TRANSPORT.token('user', 'audio', conversation_id='conversation') + with pytest.raises(HTTPException) as exc: + boundary.backup_audio_manifest(APP, 'conversation', 'Bearer ' + token) + assert exc.value.status_code in (403, 404) + assert not any(call[0] == 'audio' for call in calls) diff --git a/backend/tests/unit/test_plugin_transport_routes.py b/backend/tests/unit/test_plugin_transport_routes.py new file mode 100644 index 00000000000..2644eb7c60c --- /dev/null +++ b/backend/tests/unit/test_plugin_transport_routes.py @@ -0,0 +1,106 @@ +"""Real app HTTP metadata and real finalized-webhook composition with fake IO.""" + +import json +import os +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock +from urllib.parse import parse_qs, urlsplit + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +os.environ.setdefault('ENCRYPTION_SECRET', 'omi_ZwB2ZNqB2HHpMK6wStk7sTpavJiPTFg7gXUHnc4tFABPU6pZ2c2DKgehtfgi4RZv') + +import routers.apps as apps +import utils.app_integrations as integrations +from utils.plugin_transport import PluginTransport + +KEY = 'synthetic-transport-secret-at-least-32-bytes' +APP = 'signed-backup' +BASE = 'https://backup.example.invalid' + + +def app_record(): + return { + 'id': APP, + 'uid': 'author', + 'name': 'Backup', + 'category': 'productivity-and-organization', + 'author': 'Developer', + 'description': 'Conversation backup', + 'image': 'https://example.invalid/icon.png', + 'capabilities': ['external_integration'], + 'approved': True, + 'private': False, + 'external_integration': { + 'triggers_on': 'memory_creation', + 'webhook_url': BASE + '/conversation', + 'auth_steps': [{'name': 'Connect', 'url': BASE + '/setup/pcloud'}], + 'app_home_url': BASE + '/setup/pcloud', + 'setup_completed_url': BASE + '/setup-completed', + }, + } + + +@pytest.fixture +def config(monkeypatch): + monkeypatch.setenv('OMI_PLUGIN_TRANSPORT_JSON', json.dumps({APP: {'base_url': BASE, 'key': KEY}})) + return PluginTransport(APP, BASE, KEY) + + +def test_app_http_response_receives_a_fresh_account_bound_launch(config, monkeypatch): + record = app_record() + monkeypatch.setattr(apps, 'get_available_app_by_id_with_reviews', lambda app_id, uid: record) + monkeypatch.setattr(apps, 'get_is_user_paid_app', lambda *args: False) + api = FastAPI() + api.include_router(apps.router) + api.dependency_overrides[apps.auth.get_current_user_uid] = lambda: 'user-a' + with TestClient(api) as client: + response = client.get('/v1/apps/' + APP) + assert response.status_code == 200 + integration = response.json()['external_integration'] + for url, scope in ( + (integration['auth_steps'][0]['url'], 'setup'), + (integration['app_home_url'], 'setup'), + (integration['setup_completed_url'], 'status'), + ): + params = parse_qs(urlsplit(url).query) + assert params['uid'] == ['user-a'] + assert config.decode(params['omi_token'][0], scope)['sub'] == 'user-a' + assert 'omi_token' not in record['external_integration']['auth_steps'][0]['url'] + + +@pytest.mark.asyncio +async def test_actual_finalized_fanout_signs_payload_and_retains_idempotency(config, monkeypatch): + app = MagicMock() + app.id, app.uid, app.enabled = APP, 'author', True + app.external_integration.webhook_url = BASE + '/conversation' + app.triggers_on_conversation_creation.return_value = True + conversation = SimpleNamespace( + id='conversation-a', discarded=False, is_locked=False, source=None, client_platform='ios' + ) + monkeypatch.setattr(integrations, 'get_available_apps', lambda uid: [app]) + monkeypatch.setattr(integrations, 'is_app_webhook_disabled', lambda app_id: False) + monkeypatch.setattr( + integrations, 'conversation_to_dict', lambda value: {'id': value.id, 'structured': {'title': 'Synthetic'}} + ) + monkeypatch.setattr( + integrations, + 'safe_request_target', + lambda url: (url, {'headers': {'Host': 'backup.example.invalid'}, 'extensions': {}}), + ) + monkeypatch.setattr(integrations, 'get_webhook_circuit_breaker', lambda url: MagicMock()) + monkeypatch.setattr(integrations, 'record_app_webhook_success', lambda *args: None) + monkeypatch.setattr(integrations, 'record_app_usage', lambda *args, **kwargs: None) + client = SimpleNamespace(post=AsyncMock(return_value=SimpleNamespace(status_code=202, json=lambda: {}))) + monkeypatch.setattr(integrations, 'get_webhook_client', lambda: client) + await integrations.trigger_external_integrations( + 'user-a', conversation, idempotency_key='durable-key', require_delivery=True + ) + sent = client.post.call_args.kwargs + assert sent['headers']['X-Omi-Idempotency-Key'] == 'durable-key' + claims = config.decode(sent['headers']['X-Omi-Plugin-Authorization'], 'delivery') + assert claims['sub'] == 'user-a' and claims['path'] == '/conversation' + audio = config.decode(sent['json']['_omi_backup_audio_token'], 'audio') + assert audio['sub'] == 'user-a' and audio['conversation_id'] == 'conversation-a' diff --git a/backend/tests/unit/test_route_policy_inventory.py b/backend/tests/unit/test_route_policy_inventory.py index 8dc943392e4..90337edb5cd 100644 --- a/backend/tests/unit/test_route_policy_inventory.py +++ b/backend/tests/unit/test_route_policy_inventory.py @@ -108,8 +108,14 @@ async def ws(): assert entries[2]['observed']['timeout_class_hint'] == 'websocket' -def test_beta_promotion_token_is_registered_for_only_the_two_beta_admission_routes(): - manifest = inventory.load_manifest(inventory.DEFAULT_MANIFEST_PATH) +@pytest.fixture(scope='module') +def shipped_manifest(): + """Parse the shipped manifest once; it is large, and parsing is setup, not the behavior under test.""" + return inventory.load_manifest(inventory.DEFAULT_MANIFEST_PATH) + + +def test_beta_promotion_token_is_registered_for_only_the_two_beta_admission_routes(shipped_manifest): + manifest = shipped_manifest assert 'beta_promotion_token' in inventory.AUTH_MECHANISMS routes = [ route @@ -122,8 +128,25 @@ def test_beta_promotion_token_is_registered_for_only_the_two_beta_admission_rout ] -def test_beta_breakglass_routes_reuse_existing_admin_key_authority(): - manifest = inventory.load_manifest(inventory.DEFAULT_MANIFEST_PATH) +def test_plugin_capability_routes_declare_the_conversation_audio_scope(shipped_manifest): + manifest = shipped_manifest + routes = [ + route + for route in manifest['routes'] + if 'plugin_capability' in route.get('policy', {}).get('auth', {}).get('mechanisms', []) + ] + assert [(route['method'], route['path']) for route in routes] == [ + ('GET', '/v1/apps/{app_id}/backup-audio/{conversation_id}'), + ('HEAD', '/v1/apps/{app_id}/backup-audio/{conversation_id}'), + ] + assert all( + route['policy']['auth'] == {'mechanisms': ['plugin_capability'], 'placement': 'handler', 'scopes': ['audio']} + for route in routes + ) + + +def test_beta_breakglass_routes_reuse_existing_admin_key_authority(shipped_manifest): + manifest = shipped_manifest routes = [route for route in manifest['routes'] if route.get('path') == '/v2/desktop/beta/breakglass'] assert [(route.get('method'), route.get('path')) for route in routes] == [ ('POST', '/v2/desktop/beta/breakglass'), diff --git a/backend/utils/app_integrations.py b/backend/utils/app_integrations.py index 7e0c489f60b..caa9bb9c853 100644 --- a/backend/utils/app_integrations.py +++ b/backend/utils/app_integrations.py @@ -8,6 +8,7 @@ import time import httpx +from utils.plugin_transport import plugin_transport from utils.http_client import ( safe_request_target, @@ -308,6 +309,12 @@ async def _single(app: App): try: payload = serialize_datetimes(conversation_dict) headers = dict(pin_kwargs['headers']) + transport = plugin_transport(app.id) + if transport: + payload['_omi_backup_audio_token'] = transport.token( + uid, 'audio', ttl=7 * 86400, conversation_id=conversation.id + ) + headers.update(transport.delivery_headers(uid, url, payload)) if idempotency_key: headers['X-Omi-Idempotency-Key'] = idempotency_key async with get_webhook_semaphore(): diff --git a/backend/utils/apps.py b/backend/utils/apps.py index fcccd784f07..a0143095d5b 100644 --- a/backend/utils/apps.py +++ b/backend/utils/apps.py @@ -78,6 +78,7 @@ from utils import stripe from utils.llm.persona import condense_conversations, condense_memories, generate_persona_description, condense_tweets from utils.llm.usage_tracker import track_usage, Features +from utils.plugin_transport import plugin_transport from utils.executors import run_blocking, db_executor, llm_executor from utils.social import get_twitter_timeline from utils.marketplace_reviewers import parse_marketplace_reviewers, is_marketplace_reviewer @@ -135,7 +136,21 @@ def validate_app_endpoints_for_reenable(app_dict: Dict[str, Any], update_dict: D seen_urls: Set[str] = set() webhook_url: Any = updated_ext.get('webhook_url') or existing_ext.get('webhook_url', '') if webhook_url: - endpoints_to_check.append(('webhook', str(webhook_url), 'POST', True)) + try: + transport = plugin_transport(app_id) + if transport is not None: + transport.validate_url(str(webhook_url)) + except ValueError: + raise HTTPException( + status_code=400, + detail='Webhook does not match this app\'s trusted connection. Fix it before re-enabling.', + ) + if transport is not None: + # An authenticated delivery endpoint rejects an unsigned empty POST by + # design; its service exposes a non-mutating health probe instead. + endpoints_to_check.append(('webhook', transport.health_url(), 'GET', True)) + else: + endpoints_to_check.append(('webhook', str(webhook_url), 'POST', True)) seen_urls.add(str(webhook_url)) mcp_url: Any = updated_ext.get('mcp_server_url') or existing_ext.get('mcp_server_url', '') if mcp_url: diff --git a/backend/utils/plugin_backup_audio.py b/backend/utils/plugin_backup_audio.py new file mode 100644 index 00000000000..9395115cdb9 --- /dev/null +++ b/backend/utils/plugin_backup_audio.py @@ -0,0 +1,40 @@ +"""Scoped read of canonical saved audio, not an independent capture buffer.""" + +import jwt +from fastapi import HTTPException + +from database.conversations import get_conversation +from utils.apps import is_user_app_enabled, get_available_app_by_id +from utils.other.endpoints import enforce_account_deletion_http_access +from utils.plugin_transport import plugin_transport +from utils.sync.playback import get_audio_signed_urls + + +def backup_audio_manifest(app_id: str, conversation_id: str, authorization: str, *, include_audio: bool = True) -> dict: + try: + transport = plugin_transport(app_id) + if transport is None: + raise ValueError('Unregistered plugin') + scheme, _, token = authorization.partition(' ') + if scheme.lower() != 'bearer': + raise ValueError('Missing bearer token') + claims = transport.decode(token, 'audio') + if claims.get('conversation_id') != conversation_id: + raise ValueError('Wrong conversation') + except (ValueError, jwt.PyJWTError): + raise HTTPException(status_code=401, detail='Invalid or expired backup audio authorization') + uid = claims['sub'] + enforce_account_deletion_http_access(uid) + if not get_available_app_by_id(app_id, uid) or not is_user_app_enabled(uid, app_id): + raise HTTPException(status_code=403, detail='This backup app is no longer enabled') + conversation = get_conversation(uid, conversation_id) + if ( + not conversation + or conversation.get('discarded') + or conversation.get('is_locked') + or conversation.get('deleted') + ): + raise HTTPException(status_code=404, detail='This conversation is not available for backup') + if not include_audio: + return {'authorized': True} + return get_audio_signed_urls(uid, conversation_id, conversation.get('audio_files') or [], conversation) diff --git a/backend/utils/plugin_transport.py b/backend/utils/plugin_transport.py new file mode 100644 index 00000000000..595b14458d4 --- /dev/null +++ b/backend/utils/plugin_transport.py @@ -0,0 +1,149 @@ +"""Opt-in, per-app authenticated transport for independently deployed plugins. + +Legacy apps are unchanged. Configuration is server-only, never marketplace data. +See backend/docs/plugin-transport.md for the wire contract and rollout checklist. +""" + +import hashlib +import json +import os +import secrets +import time +from dataclasses import dataclass +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +import jwt + +ISSUER = 'omi-plugin-transport/v1' + + +@dataclass(frozen=True) +class PluginTransport: + app_id: str + base_url: str + key: str + + def validate_url(self, url: str) -> None: + expected, actual = urlsplit(self.base_url), urlsplit(url) + if ( + actual.scheme != 'https' + or actual.netloc != expected.netloc + or actual.username + or actual.password + or actual.fragment + or not actual.path.startswith(expected.path.rstrip('/') + '/') + ): + raise ValueError('Configured plugin endpoint does not match its trusted transport origin') + + def health_url(self) -> str: + """A non-mutating readiness probe on the trusted origin; deliveries stay signed.""" + return self.base_url.rstrip('/') + '/health' + + def token(self, uid: str, scope: str, *, ttl: int = 600, **claims) -> str: + now = int(time.time()) + return jwt.encode( + { + **claims, + 'iss': ISSUER, + 'aud': self.app_id, + 'sub': uid, + 'scope': scope, + 'iat': now, + 'exp': now + ttl, + 'jti': secrets.token_urlsafe(24), + }, + self.key, + algorithm='HS256', + ) + + def decode(self, token: str, scope: str) -> dict: + claims = jwt.decode( + token, + self.key, + algorithms=['HS256'], + audience=self.app_id, + issuer=ISSUER, + options={'require': ['exp', 'iat', 'sub', 'jti', 'scope']}, + ) + if claims['scope'] != scope or not isinstance(claims['sub'], str) or not claims['sub']: + raise ValueError('Invalid plugin token scope or subject') + return claims + + def setup_url(self, url: str, uid: str, scope: str = 'setup') -> str: + self.validate_url(url) + parts = urlsplit(url) + query = [(k, v) for k, v in parse_qsl(parts.query) if k not in ('uid', 'omi_token')] + query.extend([('uid', uid), ('omi_token', self.token(uid, scope))]) + return urlunsplit(parts._replace(query=urlencode(query))) + + def delivery_headers(self, uid: str, url: str, payload: dict) -> dict[str, str]: + self.validate_url(url) + digest = hashlib.sha256(canonical_payload(payload)).hexdigest() + return { + 'X-Omi-Plugin-Authorization': self.token( + uid, 'delivery', path=urlsplit(url).path, sha256=digest, method='POST' + ) + } + + +def canonical_payload(payload: dict) -> bytes: + return json.dumps(payload, sort_keys=True, separators=(',', ':'), ensure_ascii=True, allow_nan=False).encode() + + +def plugin_transport(app_id: str) -> PluginTransport | None: + """An absent registration retains legacy behavior; a malformed one fails closed.""" + raw = os.getenv('OMI_PLUGIN_TRANSPORT_JSON', '').strip() + if not raw: + return None + config = json.loads(raw) + if not isinstance(config, dict): + raise ValueError('OMI_PLUGIN_TRANSPORT_JSON must be an object') + if app_id not in config: + return None + entry = config[app_id] + if not isinstance(entry, dict): + raise ValueError('Invalid plugin transport registration') + base_url, key = entry.get('base_url'), entry.get('key') + if not isinstance(base_url, str) or not isinstance(key, str) or len(key.encode()) < 32: + raise ValueError('Plugin transport needs an HTTPS origin and at least a 32-byte key') + parts = urlsplit(base_url) + if ( + parts.scheme != 'https' + or not parts.hostname + or parts.username + or parts.password + or parts.query + or parts.fragment + ): + raise ValueError('Invalid plugin transport origin') + return PluginTransport(app_id, base_url.rstrip('/'), key) + + +def authorize_app_setup(app, uid: str): + """Decorate only this authenticated response; never modify cached app data.""" + transport = plugin_transport(app.id) + if transport is None or app.external_integration is None: + return app + result = app.model_copy(deep=True) + integration = result.external_integration + for step in integration.auth_steps or []: + step.url = transport.setup_url(step.url, uid) + if integration.app_home_url: + integration.app_home_url = transport.setup_url(integration.app_home_url, uid) + if integration.setup_completed_url: + integration.setup_completed_url = transport.setup_url(integration.setup_completed_url, uid, 'status') + return result + + +def strip_launch_credentials(app_id: str, integration: dict) -> None: + """An owner editing app details must not persist a short-lived launch grant.""" + if plugin_transport(app_id) is None: + return + values = [(integration, field) for field in ('app_home_url', 'setup_completed_url')] + values.extend((step, 'url') for step in integration.get('auth_steps') or [] if isinstance(step, dict)) + for container, field in values: + url = container.get(field) + if isinstance(url, str): + parts = urlsplit(url) + query = [(key, value) for key, value in parse_qsl(parts.query) if key not in ('uid', 'omi_token')] + container[field] = urlunsplit(parts._replace(query=urlencode(query))) diff --git a/contracts/parity/README.md b/contracts/parity/README.md index 1349093f455..d556d5cd50b 100644 --- a/contracts/parity/README.md +++ b/contracts/parity/README.md @@ -36,6 +36,7 @@ section projections, and unattributed app results. | ------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------- | | Backend/API and standalone MCP | `backend/tests/unit/test_parity_contracts.py` | Backend unit suite and Desktop Backend Contracts CI | | Flutter app | `app/test/parity/parity_contracts_test.dart` | `app/test.sh`, CI Flutter tests | +| pCloud backup plugin | `plugins/omi-pcloud-app/test_app.py` (`conversation_summary.json` only; the plugin is a standalone service, so `presentation.py` ports the backend selector) | `scripts/run-plugin-pcloud-tests.sh` | | Windows desktop | `desktop/windows/src/renderer/src/lib/parityContracts.test.ts`, `desktop/windows/src/shared/knowledgeLedger.test.ts` | `npm test` in `desktop/windows`, CI Desktop Windows tests | | macOS desktop | JIT matrix: `desktop/macos/Desktop/Tests/ServerMemoryV17DecodingTests.swift`. Duration: `desktop/macos/Desktop/Tests/ConversationDurationTests.swift`. Task/day adapter and the capture-group collapse adapter (`CaptureGroupPresentationTests.swift` pins the same cases inline) remain pending. | Desktop Swift CI | | Web app | `web/app/src/lib/__tests__/knowledgeLedger.test.ts` | `web/app/test.sh`, CI Web App checks | diff --git a/desktop/macos/Desktop/Sources/MainWindow/Pages/AppsPage.swift b/desktop/macos/Desktop/Sources/MainWindow/Pages/AppsPage.swift index aca18f3bb47..9490b87c327 100644 --- a/desktop/macos/Desktop/Sources/MainWindow/Pages/AppsPage.swift +++ b/desktop/macos/Desktop/Sources/MainWindow/Pages/AppsPage.swift @@ -2781,6 +2781,7 @@ struct AppDetailSheet: View { @State private var showAddReview = false @State private var userReview: OmiAppReview? @State private var appDetails: OmiAppDetails? + @State private var isDetailVisible = true @State private var isSettingUp = false @State private var isSetupCompleted = false @State private var setupCheckTask: Task? @@ -2901,7 +2902,7 @@ struct AppDetailSheet: View { switch primaryAction { case .open: // Open the external integration in browser - openExternalApp() + await openExternalApp() case .install: if app.worksExternally { await handleInstall() @@ -2966,11 +2967,7 @@ struct AppDetailSheet: View { VStack(alignment: .leading, spacing: OmiSpacing.sm) { ForEach(Array(integration.authSteps.enumerated()), id: \.offset) { index, step in Button(action: { - if let uid = AuthState.shared.userId, - let url = AppSetupURL.withUID(step.url, uid: uid) - { - NSWorkspace.shared.open(url) - } + Task { await navigateToSetup(stepIndex: index) } }) { HStack(spacing: OmiSpacing.md) { // Step number / checkmark @@ -3107,12 +3104,14 @@ struct AppDetailSheet: View { .frame(width: 500, height: 600) .background(Ink.surface) .task { + isDetailVisible = true await loadReviews() await loadAppDetails() // Resume polling if user completed setup in browser and returned to this sheet await resumeSetupPollingIfNeeded() } .onDisappear { + isDetailVisible = false setupCheckTask?.cancel() } .dismissableSheet(isPresented: $showAddReview) { @@ -3185,9 +3184,12 @@ struct AppDetailSheet: View { startSetupPolling(completionUrl: completionUrl, uid: uid) } - private func openExternalApp() { + /// A transport-backed app's home URL carries a one-use, short-lived launch + /// grant, so details are fetched immediately before each open, never reused. + private func openExternalApp() async { + guard let details = await freshLaunchDetails() else { return } guard let uid = AuthState.shared.userId else { return } - let integration = appDetails?.externalIntegration + let integration = details.externalIntegration // Prefer appHomeUrl, then first auth step URL if let homeUrl = integration?.appHomeUrl, !homeUrl.isEmpty, let url = URL(string: homeUrl) { NSWorkspace.shared.open(url) @@ -3206,18 +3208,27 @@ struct AppDetailSheet: View { if isEnabled { return } // Step 3: Enable failed — app requires setup first. Open browser and wait. - // Ensure app details are loaded before navigating to setup. - if appDetails == nil { await loadAppDetails() } await navigateToSetup() } - private func navigateToSetup() async { + private func freshLaunchDetails() async -> OmiAppDetails? { + let details = await AppSetupURL.freshDetails( + currentUID: { isDetailVisible ? AuthState.shared.userId : nil }, + load: { try await APIClient.shared.getAppDetails(appId: app.id) } + ) + if let details { appDetails = details } + return details + } + + private func navigateToSetup(stepIndex: Int = 0) async { + guard let details = await freshLaunchDetails() else { return } guard let uid = AuthState.shared.userId else { return } - let integration = appDetails?.externalIntegration + let integration = details.externalIntegration // Open auth step or setup instructions URL in browser if let authSteps = integration?.authSteps, !authSteps.isEmpty { - if let url = AppSetupURL.withUID(authSteps[0].url, uid: uid) { + guard authSteps.indices.contains(stepIndex) else { return } + if let url = AppSetupURL.withUID(authSteps[stepIndex].url, uid: uid) { NSWorkspace.shared.open(url) } } else if let instructionsPath = integration?.setupInstructionsFilePath, !instructionsPath.isEmpty { diff --git a/desktop/macos/Desktop/Sources/Services/APIClient/AppSetupURL.swift b/desktop/macos/Desktop/Sources/Services/APIClient/AppSetupURL.swift index 5dca93e2bf9..e7d01dafe4d 100644 --- a/desktop/macos/Desktop/Sources/Services/APIClient/AppSetupURL.swift +++ b/desktop/macos/Desktop/Sources/Services/APIClient/AppSetupURL.swift @@ -16,6 +16,22 @@ import Foundation /// signature depends on. Any `uid` the base already carries is dropped so the /// handed-off value is unambiguous. enum AppSetupURL { + /// A launch capability belongs to one fresh read and its original account. + /// Failure or an account change must never fall back to cached credentials. + @MainActor + static func freshDetails( + currentUID: () -> String?, load: @MainActor () async throws -> Value + ) async -> Value? { + guard let uid = currentUID() else { return nil } + do { + let value = try await load() + guard !Task.isCancelled, currentUID() == uid else { return nil } + return value + } catch { + return nil + } + } + static func withUID(_ base: String, uid: String) -> URL? { guard !base.isEmpty, var components = URLComponents(string: base) else { return nil } var uidComponents = URLComponents() diff --git a/desktop/macos/Desktop/Tests/AppSetupURLTests.swift b/desktop/macos/Desktop/Tests/AppSetupURLTests.swift index 96fc3092f9f..46d7b553760 100644 --- a/desktop/macos/Desktop/Tests/AppSetupURLTests.swift +++ b/desktop/macos/Desktop/Tests/AppSetupURLTests.swift @@ -73,4 +73,34 @@ final class AppSetupURLTests: XCTestCase { func testInvalidBaseResolvesNil() { XCTAssertNil(AppSetupURL.withUID("", uid: "u1")) } + + @MainActor + func testEachLaunchFetchesFreshDetails() async { + var reads = 0 + for expected in 1...2 { + let value = await AppSetupURL.freshDetails(currentUID: { "u1" }) { + reads += 1 + return "signed.\(reads)" + } + XCTAssertEqual(value, "signed.\(expected)") + } + } + + @MainActor + func testFailedRefreshCannotReturnCachedCredentials() async { + let value: String? = await AppSetupURL.freshDetails(currentUID: { "u1" }) { + throw URLError(.notConnectedToInternet) + } + XCTAssertNil(value) + } + + @MainActor + func testAccountReplacementDiscardsFreshCredentials() async { + var uid: String? = "u1" + let value = await AppSetupURL.freshDetails(currentUID: { uid }) { + uid = "u2" + return "signed.for.u1" + } + XCTAssertNil(value) + } } diff --git a/desktop/macos/changelog/unreleased/20261021-app-home-fresh-launch-grant.json b/desktop/macos/changelog/unreleased/20261021-app-home-fresh-launch-grant.json new file mode 100644 index 00000000000..3d079d5de63 --- /dev/null +++ b/desktop/macos/changelog/unreleased/20261021-app-home-fresh-launch-grant.json @@ -0,0 +1,3 @@ +{ + "change": "Opening app settings or retrying an integration setup now uses a fresh sign-in link instead of an expired or already-used link" +} diff --git a/desktop/windows/changelog/unreleased/20261021-app-home-fresh-launch-grant.json b/desktop/windows/changelog/unreleased/20261021-app-home-fresh-launch-grant.json new file mode 100644 index 00000000000..e2ae4da293c --- /dev/null +++ b/desktop/windows/changelog/unreleased/20261021-app-home-fresh-launch-grant.json @@ -0,0 +1,5 @@ +{ + "changes": [ + "Opening app settings or retrying an integration setup now uses a fresh sign-in link instead of an expired or already-used link" + ] +} diff --git a/desktop/windows/changelog/unreleased/20261023-apps-install-fresh-setup-grant.json b/desktop/windows/changelog/unreleased/20261023-apps-install-fresh-setup-grant.json new file mode 100644 index 00000000000..ee95d2e82f5 --- /dev/null +++ b/desktop/windows/changelog/unreleased/20261023-apps-install-fresh-setup-grant.json @@ -0,0 +1,5 @@ +{ + "changes": [ + "Installing an integration for the first time from the Apps page now opens a fresh sign-in link for your account, and stops cleanly if the link cannot be loaded or you switch accounts" + ] +} diff --git a/desktop/windows/src/renderer/src/components/apps/AppDetailSheet.test.tsx b/desktop/windows/src/renderer/src/components/apps/AppDetailSheet.test.tsx index a86cd889639..365bd60f424 100644 --- a/desktop/windows/src/renderer/src/components/apps/AppDetailSheet.test.tsx +++ b/desktop/windows/src/renderer/src/components/apps/AppDetailSheet.test.tsx @@ -129,10 +129,71 @@ describe('AppDetailSheet layout', () => { await waitFor(() => expect(screen.getByText('Connect account')).toBeTruthy()) expect(screen.getByText('Click to complete')).toBeTruthy() fireEvent.click(screen.getByText('Connect account')) - expect( - (window as unknown as { omi: { openExternalUrl: ReturnType } }).omi - .openExternalUrl - ).toHaveBeenCalledWith('https://ex.com/setup?uid=me') + await waitFor(() => + expect( + (window as unknown as { omi: { openExternalUrl: ReturnType } }).omi + .openExternalUrl + ).toHaveBeenCalledWith('https://ex.com/setup?uid=me') + ) + }) + + it('refetches setup grants on each click without corrupting a signed query', async () => { + let grant = 0 + omiGet.mockImplementation((url: string) => { + if (url.endsWith('/reviews')) return Promise.resolve({ data: [] }) + grant += 1 + return Promise.resolve({ + data: { + ...DETAIL, + external_integration: { + auth_steps: [ + { + name: 'Connect account', + url: `https://ex.com/setup?uid=me&omi_token=signed.${grant}` + } + ] + } + } + }) + }) + renderSheet() + const step = await screen.findByText('Connect account') + fireEvent.click(step) + await waitFor(() => + expect(openSpy()).toHaveBeenLastCalledWith('https://ex.com/setup?omi_token=signed.2&uid=me') + ) + fireEvent.click(screen.getByText('Connect account')) + await waitFor(() => + expect(openSpy()).toHaveBeenLastCalledWith('https://ex.com/setup?omi_token=signed.3&uid=me') + ) + }) + + it('does not reuse the cached setup grant when refresh fails', async () => { + mockGets(DETAIL, []) + renderSheet() + const step = await screen.findByText('Connect account') + omiGet.mockRejectedValueOnce(new Error('offline')) + fireEvent.click(step) + await waitFor(() => expect(toastFn).toHaveBeenCalled()) + expect(openSpy()).not.toHaveBeenCalled() + }) + + it('discards a launch if the account changes while details load', async () => { + mockGets(DETAIL, []) + renderSheet() + const step = await screen.findByText('Connect account') + let finish!: (value: { data: App }) => void + omiGet.mockImplementationOnce( + () => + new Promise((resolve) => { + finish = resolve + }) + ) + fireEvent.click(step) + getCacheUidMock.mockReturnValue('another-account') + finish({ data: DETAIL }) + await Promise.resolve() + expect(openSpy()).not.toHaveBeenCalled() }) it('omits the Setup section when there are no auth steps', async () => { @@ -234,7 +295,7 @@ describe('AppDetailSheet open installed (external integrations)', () => { renderSheet({ enabled: true }) const btn = await screen.findByRole('button', { name: /^Open$/ }) fireEvent.click(btn) - expect(openSpy()).toHaveBeenCalledWith('https://home.example.com/app') + await waitFor(() => expect(openSpy()).toHaveBeenCalledWith('https://home.example.com/app')) // Disable is still reachable via the trash affordance. expect(screen.getByLabelText('Disable app')).toBeTruthy() }) @@ -252,7 +313,31 @@ describe('AppDetailSheet open installed (external integrations)', () => { renderSheet({ enabled: true }) const btn = await screen.findByRole('button', { name: /^Open$/ }) fireEvent.click(btn) - expect(openSpy()).toHaveBeenCalledWith('https://ex.com/setup?uid=me') + await waitFor(() => expect(openSpy()).toHaveBeenCalledWith('https://ex.com/setup?uid=me')) + }) + + it('fetches a fresh launch grant on each Open instead of reusing the loaded one', async () => { + let grant = 0 + omiGet.mockImplementation((url: string) => { + if (url.endsWith('/reviews')) return Promise.resolve({ data: [] }) + grant += 1 + return Promise.resolve({ + data: { + ...DETAIL, + external_integration: { app_home_url: `https://home.example.com/app?grant=${grant}` } + } + }) + }) + renderSheet({ enabled: true }) + const btn = await screen.findByRole('button', { name: /^Open$/ }) + fireEvent.click(btn) + await waitFor(() => + expect(openSpy()).toHaveBeenCalledWith('https://home.example.com/app?grant=2') + ) + fireEvent.click(btn) + await waitFor(() => + expect(openSpy()).toHaveBeenCalledWith('https://home.example.com/app?grant=3') + ) }) it('with no signed-in uid the auth-step fallback is guarded (no-op)', async () => { diff --git a/desktop/windows/src/renderer/src/components/apps/AppDetailSheet.tsx b/desktop/windows/src/renderer/src/components/apps/AppDetailSheet.tsx index e40d4a20a81..869b46c63e3 100644 --- a/desktop/windows/src/renderer/src/components/apps/AppDetailSheet.tsx +++ b/desktop/windows/src/renderer/src/components/apps/AppDetailSheet.tsx @@ -14,7 +14,7 @@ import { import { omiApi } from '../../lib/apiClient' import { getCacheUid } from '../../lib/persistentCache' import { toast } from '../../lib/toast' -import { startSetupPolling, worksExternally } from '../../lib/appInstall' +import { startSetupPolling, urlWithUidParam, worksExternally } from '../../lib/appInstall' import type { App, AppCatalogItem, AppReview } from '../../lib/omiApi.generated' import { AddReviewDialog } from './AddReviewDialog' @@ -287,14 +287,32 @@ export function AppDetailSheet({ // app_home_url, else the first auth step carrying the user's uid. The auth-step // fallback needs a signed-in uid; when there is neither a home url nor a resolvable // step, do nothing (no dead link). - const onOpenApp = (): void => { - const homeUrl = integration?.app_home_url - const target = - homeUrl || (currentUid && authSteps[0]?.url ? `${authSteps[0].url}?uid=${currentUid}` : null) - if (!target) return - void window.omi.openExternalUrl(target).then((ok) => { - if (!ok) toast("This app's link is unavailable.", { tone: 'warn' }) - }) + // A transport-backed app's home url carries a one-use, short-lived launch grant, + // so every home/setup open refetches. Never reuse a grant after a failed read. + const onOpenApp = (stepIndex?: number): void => { + void (async () => { + const uid = getCacheUid() + try { + const res = await omiApi.get(`/v1/apps/${app.id}`) + if (!mounted.current || getCacheUid() !== uid) return + setDetail(res.data) + const fresh = res.data.external_integration + const step = fresh?.auth_steps?.[stepIndex ?? 0] + const target = + stepIndex === undefined && fresh?.app_home_url + ? fresh.app_home_url + : uid && step?.url + ? urlWithUidParam(step.url, uid) + : null + if (!target) return + const ok = await window.omi.openExternalUrl(target) + if (!ok) toast("This app's link is unavailable.", { tone: 'warn' }) + } catch { + if (mounted.current && getCacheUid() === uid) { + toast('Could not refresh this app. Try again.', { tone: 'warn' }) + } + } + })() } // Primary button states: "Setting up…" while polling an external setup, "Installed" @@ -370,7 +388,7 @@ export function AppDetailSheet({ ) : enabled ? ( external ? ( ' + else: + action = ( + f'
{token}
' + ) + rows = [] + for job in jobs: + count = len(job.get('uploaded', [])) + file_label = 'file' if count == 1 else 'files' + retry = '' + if job['state'] in ('failed', 'pending'): + retry = f'
{token}
' + error = f'

{escape(job.get("error") or "")}

' if job.get('error') else '' + rows.append( + f'
  • {escape(job.get("title") or "Conversation")}

    {escape(job["state"].replace("_", " ").capitalize())} · {count} {file_label} saved

    {error}{retry}
    • ' + ) + rows_html = ''.join(rows) or '
  • No backups yet. New conversations appear here after setup is complete.
    • ' + choices = ''.join( + f'' + for key, label in [ + ('save_summary', 'Conversation summaries'), + ('save_transcript', 'Private transcripts — send a copy to pCloud'), + ('save_audio', 'Saved conversation audio — send a copy to pCloud'), + ] + ) + connection_text = ( + f'Connected as {escape(connection.get("email") or "your pCloud account")}' + if status == 'active' + else ('Backups stopped. pCloud access revocation is still pending.' if connection else 'Not connected') + ) + notice_html = f'

    {escape(notice)}

    ' if notice else '' + return f'''pCloud backup settings · Omi

    pCloud backups

    {notice_html}

    {connection_text}

    {action}

    Choose what to back up

    {token}
    Backup preferences{choices}

    Transcripts and audio stay off until you explicitly enable and save them. Audio is available only when Omi has saved it for that conversation.

    Recent backups

    Failed backups retry automatically up to five times. You can also retry below. Pending content expires after seven days.

      {rows_html}

    Disconnecting and privacy

    Disconnect stops new backups and cancels pending copies. Omi removes local credentials after pCloud confirms revocation; if pCloud is unavailable, encrypted credentials are retained only so you can retry revocation. Omi does not delete files already saved to pCloud.

    Open these settings from Omi on either mobile or desktop to see the same saved connection and backup status.
    ''' diff --git a/plugins/omi-pcloud-app/provider_contract.py b/plugins/omi-pcloud-app/provider_contract.py new file mode 100644 index 00000000000..7a633e4eff5 --- /dev/null +++ b/plugins/omi-pcloud-app/provider_contract.py @@ -0,0 +1,51 @@ +""" +Protocol contract for extensible cloud backup providers in Omi. +Enables pluggable destinations (pCloud, Dropbox, S3, WebDAV). +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional, Protocol, Tuple, runtime_checkable + + +@dataclass(frozen=True) +class BackupUploadResult: + """Unified result metadata returned by cloud backup providers.""" + + file_id: str + filename: str + path: str + size_bytes: int + modified_time: str + + +@runtime_checkable +class CloudBackupProvider(Protocol): + """Formal protocol for cloud backup destinations.""" + + provider_id: str + + def ensure_folder(self, folder_path: str) -> Tuple[Optional[str | int], Optional[str]]: + """Ensures the destination folder hierarchy exists. + + Returns (folder_identifier, error_message). + """ + ... + + def upload_file( + self, + folder_ref: str | int, + filename: str, + content: bytes, + overwrite: bool = True, + ) -> Tuple[Optional[BackupUploadResult], Optional[str]]: + """Uploads binary file content to the destination folder. + + Returns (BackupUploadResult, error_message). + """ + ... + + def sanitize_path(self, name: str) -> str: + """Normalizes and strips invalid path characters.""" + ... diff --git a/plugins/omi-pcloud-app/railway.toml b/plugins/omi-pcloud-app/railway.toml new file mode 100644 index 00000000000..75d233e6797 --- /dev/null +++ b/plugins/omi-pcloud-app/railway.toml @@ -0,0 +1,9 @@ +[build] +builder = "NIXPACKS" + +[deploy] +startCommand = "python -m uvicorn main:app --host 0.0.0.0 --port $PORT --no-access-log" +healthcheckPath = "/health" +healthcheckTimeout = 300 +restartPolicyType = "ON_FAILURE" +restartPolicyMaxRetries = 10 diff --git a/plugins/omi-pcloud-app/requirements.txt b/plugins/omi-pcloud-app/requirements.txt new file mode 100644 index 00000000000..a1c969552b4 --- /dev/null +++ b/plugins/omi-pcloud-app/requirements.txt @@ -0,0 +1,9 @@ +fastapi==0.121.0 +uvicorn==0.30.5 +python-multipart==0.0.31 +requests==2.33.1 +requests-toolbelt==1.0.0 +pydantic==2.13.4 +cryptography==48.0.1 +PyJWT==2.13.0 +../omi-plugin-sdk diff --git a/plugins/omi-pcloud-app/security.py b/plugins/omi-pcloud-app/security.py new file mode 100644 index 00000000000..a40fdd67149 --- /dev/null +++ b/plugins/omi-pcloud-app/security.py @@ -0,0 +1,51 @@ +"""Verify backend-issued, account-bound grants; UID hints never authorize.""" + +import hashlib +import json +import os + +import jwt +from fastapi import HTTPException, Request + +ISSUER = 'omi-plugin-transport/v1' + + +def signing_key() -> str: + key = os.getenv('OMI_PLUGIN_SIGNING_KEY', '') + if len(key.encode()) < 32 or not os.getenv('OMI_PLUGIN_APP_ID'): + raise ValueError('Configure OMI_PLUGIN_APP_ID and a 32-byte OMI_PLUGIN_SIGNING_KEY') + return key + + +def verify(token: str, scope: str) -> dict: + try: + claims = jwt.decode( + token, + signing_key(), + algorithms=['HS256'], + audience=os.environ['OMI_PLUGIN_APP_ID'], + issuer=ISSUER, + options={'require': ['exp', 'iat', 'sub', 'jti', 'scope']}, + ) + if claims['scope'] != scope or not isinstance(claims['sub'], str) or not claims['sub']: + raise ValueError('Wrong scope') + return claims + except ValueError: + raise HTTPException(status_code=401, detail='Open pCloud settings again from your signed-in Omi app.') + except jwt.PyJWTError: + raise HTTPException( + status_code=401, detail='This connection link is invalid or expired. Open it again from Omi.' + ) + + +def verify_delivery(request: Request, payload: dict, uid: str) -> dict: + claims = verify(request.headers.get('X-Omi-Plugin-Authorization', ''), 'delivery') + canonical = json.dumps(payload, sort_keys=True, separators=(',', ':'), ensure_ascii=True, allow_nan=False).encode() + if ( + claims['sub'] != uid + or claims.get('path') != request.url.path + or claims.get('method') != 'POST' + or claims.get('sha256') != hashlib.sha256(canonical).hexdigest() + ): + raise HTTPException(status_code=401, detail='The backup delivery identity or content did not match.') + return claims diff --git a/plugins/omi-pcloud-app/store.py b/plugins/omi-pcloud-app/store.py new file mode 100644 index 00000000000..a5c2c94f343 --- /dev/null +++ b/plugins/omi-pcloud-app/store.py @@ -0,0 +1,360 @@ +"""Encrypted durable state for the standalone, persistent-volume deployment. + +SQLite transactions own nonce consumption, job admission and worker leases. +Per-account OS locks serialize settings/disconnect with each provider operation. +There is no automatic fallback from a failed production store to local files. +""" + +from __future__ import annotations + +import fcntl +import hashlib +import json +import os +import secrets +import sqlite3 +import time +from contextlib import contextmanager +from pathlib import Path + +from cryptography.fernet import Fernet, MultiFernet, InvalidToken + +DEFAULT_SETTINGS = { + 'folder_name': 'Omi Conversations', + 'save_summary': True, + 'save_transcript': False, + 'save_audio': False, + 'consent_version': 0, +} +JOB_RETENTION_SECONDS = 7 * 86400 + + +class Store: + def __init__(self, path: str, keys: list[str]): + if not Path(path).is_absolute() or not keys: + raise ValueError('Configure an absolute PCLOUD_DB_PATH and PCLOUD_ENCRYPTION_KEYS') + self.path = Path(path) + self.cipher = MultiFernet([Fernet(key.encode()) for key in keys]) + + @contextmanager + def transaction(self): + self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + conn = sqlite3.connect(str(self.path), timeout=5, isolation_level=None) + os.chmod(self.path, 0o600) + try: + conn.execute('PRAGMA busy_timeout=5000') + conn.execute('PRAGMA synchronous=FULL') + conn.execute( + 'CREATE TABLE IF NOT EXISTS records (kind TEXT, id TEXT, uid TEXT, blob BLOB, expires REAL, PRIMARY KEY(kind,id))' + ) + conn.execute( + 'CREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, uid TEXT, blob BLOB, state TEXT, due REAL, lease TEXT, expires REAL)' + ) + conn.execute('BEGIN IMMEDIATE') + yield conn + conn.commit() + except BaseException: + conn.rollback() + raise + finally: + conn.close() + + def seal(self, uid: str, data: dict) -> bytes: + return self.cipher.encrypt(json.dumps({'uid': uid, 'data': data}, ensure_ascii=True).encode()) + + def open(self, uid: str, blob: bytes) -> dict: + envelope = json.loads(self.cipher.decrypt(blob)) + if envelope.get('uid') != uid: + raise ValueError('Stored principal does not match record owner') + return envelope['data'] + + @contextmanager + def account_lock(self, uid: str): + self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + digest = hashlib.sha256(uid.encode()).hexdigest() + with (self.path.parent / f'.account-{digest}.lock').open('a+b') as handle: + fcntl.flock(handle, fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(handle, fcntl.LOCK_UN) + + def get(self, kind: str, identity: str, *, consume: bool = False) -> dict | None: + with self.transaction() as conn: + row = conn.execute( + 'SELECT uid,blob,expires FROM records WHERE kind=? AND id=?', + (kind, identity), + ).fetchone() + if row is None: + return None + if consume or row[2] <= time.time(): + conn.execute('DELETE FROM records WHERE kind=? AND id=?', (kind, identity)) + if row[2] <= time.time(): + return None + return self.open(row[0], row[1]) + + def put( + self, + kind: str, + identity: str, + uid: str, + data: dict, + ttl: int = JOB_RETENTION_SECONDS, + ): + with self.transaction() as conn: + conn.execute( + 'INSERT OR REPLACE INTO records VALUES (?,?,?,?,?)', + (kind, identity, uid, self.seal(uid, data), time.time() + ttl), + ) + + def delete(self, kind: str, identity: str): + with self.transaction() as conn: + conn.execute('DELETE FROM records WHERE kind=? AND id=?', (kind, identity)) + + def consume_nonce(self, identity: str, uid: str, ttl: int) -> bool: + with self.transaction() as conn: + conn.execute('DELETE FROM records WHERE expires<=?', (time.time(),)) + try: + conn.execute( + 'INSERT INTO records VALUES (?,?,?,?,?)', + ('used', identity, uid, self.seal(uid, {}), time.time() + ttl), + ) + except sqlite3.IntegrityError: + return False + return True + + def settings(self, uid: str) -> dict: + settings = {**DEFAULT_SETTINGS, **(self.get('settings', uid) or {})} + # Old/missing consent versions are never treated as a transcript/audio grant. + if settings.get('consent_version') != 1: + settings.update(save_transcript=False, save_audio=False) + return settings + + def check_ready(self): + with self.transaction() as conn: + row = conn.execute( + 'SELECT uid,blob FROM records WHERE kind=? AND id=?', + ('meta', 'encryption-canary'), + ).fetchone() + if row is None: + sample = conn.execute('SELECT uid,blob FROM records LIMIT 1').fetchone() + if sample: + self.open(sample[0], sample[1]) + conn.execute( + 'INSERT INTO records VALUES (?,?,?,?,?)', + ( + 'meta', + 'encryption-canary', + '__service__', + self.seal('__service__', {'schema': 1}), + time.time() + 10 * 365 * 86400, + ), + ) + else: + self.open(row[0], row[1]) + + # An expired or canceled backup is a scrubbed tombstone, not a delivery to + # deduplicate: a fresh authenticated delivery (re-export, or reconnect after + # disconnect) replaces it. Active, failed, completed and unreadable rows keep + # deduplicating so a backup is never uploaded twice or a restore is lost. + REPLACEABLE_JOB_STATES = ('expired', 'canceled') + + def enqueue(self, uid: str, conversation_id: str, payload: dict) -> str: + identity = hashlib.sha256(f'{uid}\0{conversation_id}'.encode()).hexdigest() + with self.transaction() as conn: + row = conn.execute('SELECT state FROM jobs WHERE id=?', (identity,)).fetchone() + if row and row[0] not in self.REPLACEABLE_JOB_STATES: + return identity + count = conn.execute( + 'SELECT COUNT(*) FROM jobs WHERE uid=? AND state NOT IN (?,?,?)', + (uid, 'completed', 'canceled', 'expired'), + ).fetchone()[0] + if count >= 100: + raise OverflowError('Too many unfinished backups. Review failed backups before trying again.') + payload.update(id=identity, attempt=0, uploaded=[], error=None) + now = time.time() + if row: + conn.execute( + 'UPDATE jobs SET blob=?,state=?,due=?,lease=?,expires=? WHERE id=?', + (self.seal(uid, payload), 'pending', now, '', now + JOB_RETENTION_SECONDS, identity), + ) + return identity + conn.execute( + 'INSERT INTO jobs VALUES (?,?,?,?,?,?,?)', + ( + identity, + uid, + self.seal(uid, payload), + 'pending', + now, + '', + now + JOB_RETENTION_SECONDS, + ), + ) + return identity + + def jobs(self, uid: str) -> list[dict]: + # The 30 most recent jobs, plus every older one the owner can still act on. + # Unfinished jobs are capped at 100 per account, so this stays bounded. + with self.transaction() as conn: + rows = conn.execute( + 'SELECT id,blob,state,expires FROM jobs WHERE uid=? AND (' + 'rowid IN (SELECT rowid FROM jobs WHERE uid=? ORDER BY rowid DESC LIMIT 30) ' + 'OR state IN (?,?)) ORDER BY rowid DESC', + (uid, uid, 'failed', 'pending'), + ).fetchall() + result = [] + for identity, blob, state, expires in rows: + try: + payload = self.open(uid, blob) + except (InvalidToken, ValueError, TypeError): + payload = { + 'uploaded': [], + 'error': 'Stored backup cannot be read. Contact the app administrator to restore it from backup.', + } + state = 'unreadable' + result.append({**payload, 'id': identity, 'state': state, 'expires': expires}) + return result + + def job_state(self, uid: str, identity: str) -> str: + with self.transaction() as conn: + row = conn.execute('SELECT state FROM jobs WHERE uid=? AND id=?', (uid, identity)).fetchone() + return row[0] if row else 'missing' + + def claim(self) -> tuple[str, str, str, dict] | None: + with self.transaction() as conn: + now = time.time() + conn.execute('DELETE FROM records WHERE expires<=?', (now,)) + conn.execute('DELETE FROM jobs WHERE expires? ORDER BY due LIMIT 1', + ('pending', 'running', now, now), + ).fetchone() + if row is None: + return None + identity, uid, blob = row + try: + payload = self.open(uid, blob) + except (InvalidToken, ValueError, TypeError): + # Preserve the original ciphertext for restore; do not strand + # unrelated accounts behind one unreadable queue head. + conn.execute('UPDATE jobs SET state=? WHERE id=?', ('unreadable', identity)) + return None + lease = secrets.token_urlsafe(24) + conn.execute( + 'UPDATE jobs SET state=?,due=?,lease=? WHERE id=?', + ('running', now + 300, lease, identity), + ) + return identity, uid, lease, payload + + def settle( + self, + identity: str, + uid: str, + lease: str, + payload: dict, + state: str, + delay: int = 0, + ): + with self.transaction() as conn: + changed = conn.execute( + 'UPDATE jobs SET blob=?,state=?,due=? WHERE id=? AND uid=? AND lease=?', + ( + self.seal(uid, payload), + state, + time.time() + delay, + identity, + uid, + lease, + ), + ).rowcount + if changed != 1: + raise RuntimeError('Backup worker lease is no longer current') + + def retry(self, uid: str, identity: str) -> bool: + with self.transaction() as conn: + row = conn.execute( + 'SELECT blob,state,expires FROM jobs WHERE uid=? AND id=?', + (uid, identity), + ).fetchone() + if not row or row[1] not in ('failed', 'pending') or row[2] <= time.time(): + return False + payload = self.open(uid, row[0]) + payload.update(attempt=0, error=None) + conn.execute( + 'UPDATE jobs SET blob=?,state=?,due=?,lease=? WHERE uid=? AND id=?', + (self.seal(uid, payload), 'pending', time.time(), '', uid, identity), + ) + return True + + def cancel_jobs(self, uid: str): + with self.transaction() as conn: + rows = conn.execute( + 'SELECT id,blob FROM jobs WHERE uid=? AND state NOT IN (?,?,?)', + (uid, 'completed', 'canceled', 'expired'), + ).fetchall() + for identity, blob in rows: + try: + previous = self.open(uid, blob) + except (InvalidToken, ValueError, TypeError, KeyError): + previous = {} + # Keep truthful receipts, not private export payloads or tokens. + receipt = { + 'id': identity, + 'title': previous.get('title', 'Conversation'), + 'uploaded': previous.get('uploaded', []), + 'error': 'Backup canceled on disconnect.', + } + conn.execute( + 'UPDATE jobs SET state=?,blob=?,lease=? WHERE id=?', + ( + 'canceled', + self.seal(uid, receipt), + '', + identity, + ), + ) + + def rotate_keys(self): + """Rewrap all ciphertext with the first configured key, atomically.""" + with self.transaction() as conn: + for kind, identity, blob in conn.execute('SELECT kind,id,blob FROM records').fetchall(): + conn.execute( + 'UPDATE records SET blob=? WHERE kind=? AND id=?', + (self.cipher.rotate(blob), kind, identity), + ) + for identity, blob in conn.execute('SELECT id,blob FROM jobs').fetchall(): + conn.execute( + 'UPDATE jobs SET blob=? WHERE id=?', + (self.cipher.rotate(blob), identity), + ) + + +def configured_store() -> Store: + return Store( + os.environ.get('PCLOUD_DB_PATH', ''), + json.loads(os.environ.get('PCLOUD_ENCRYPTION_KEYS', '[]')), + ) diff --git a/plugins/omi-pcloud-app/test_app.py b/plugins/omi-pcloud-app/test_app.py new file mode 100644 index 00000000000..970bab96cb0 --- /dev/null +++ b/plugins/omi-pcloud-app/test_app.py @@ -0,0 +1,879 @@ +"""Real HTTP/SQLite regressions for #20498 and #20213; provider network is denied. + +Expected consent semantics come from #20213. Region/revocation expectations come +from docs.pcloud.com/methods/oauth_2.0/authorize.html and methods/auth/logout.html. +""" + +import copy +import concurrent.futures +import hashlib +import json +import os +import secrets +import sys +import time +from pathlib import Path +from types import SimpleNamespace +from urllib.parse import parse_qs, urlsplit + +import jwt +import pytest +import requests +from cryptography.fernet import Fernet, MultiFernet, InvalidToken +from fastapi.testclient import TestClient + +sys.path.append(str(Path(__file__).resolve().parents[2] / 'backend')) +from utils.plugin_transport import PluginTransport, authorize_app_setup + +import main +import worker +from pcloud_client import PCloudClient +from provider_contract import BackupUploadResult +from store import Store + +KEY = 'synthetic-test-key-at-least-32-bytes-long' +APP = 'synthetic-pcloud-app' +UID = 'synthetic-user-a' +TRANSPORT = PluginTransport(APP, 'https://pcloud.example.invalid', KEY) + + +def payload(identity='conversation-a', title='Same title'): + return { + 'id': identity, + 'created_at': '2026-10-20T10:00:00Z', + 'finished_at': '2026-10-20T10:30:00Z', + 'structured': { + 'title': title, + 'overview': 'Synthetic summary', + 'category': 'work', + }, + 'transcript_segments': [ + { + 'text': 'Synthetic private transcript', + 'is_user': True, + 'start': 0, + 'end': 1, + } + ], + 'discarded': False, + '_omi_backup_audio_token': TRANSPORT.token(UID, 'audio', conversation_id=identity, ttl=86400), + } + + +@pytest.fixture +def env(tmp_path, monkeypatch): + monkeypatch.setenv('OMI_PLUGIN_SIGNING_KEY', KEY) + monkeypatch.setenv('OMI_PLUGIN_APP_ID', APP) + monkeypatch.setenv('PCLOUD_CLIENT_ID', 'synthetic-client') + monkeypatch.setenv('PCLOUD_CLIENT_SECRET', 'synthetic-secret') + monkeypatch.setenv('PCLOUD_REDIRECT_URI', 'https://pcloud.example.invalid/auth/pcloud/callback') + key = Fernet.generate_key().decode() + store = Store(str(tmp_path / 'pcloud.sqlite3'), [key]) + monkeypatch.setattr(main, '_store', store) + monkeypatch.setattr( + requests.sessions.Session, + 'request', + lambda *a, **kw: (_ for _ in ()).throw(AssertionError('Live network forbidden')), + ) + monkeypatch.setattr(requests, 'head', lambda *args, **kwargs: SimpleNamespace(status_code=200)) + # No lifespan worker in route tests: drive its real production function explicitly. + client = TestClient(main.app, base_url='https://pcloud.example.invalid') + result = SimpleNamespace(store=store, client=client, monkeypatch=monkeypatch, key=key) + yield result + client.close() + + +def login(env, uid=UID, client=None): + client = client or env.client + result = client.get( + '/setup/pcloud', + params={'uid': uid, 'omi_token': TRANSPORT.token(uid, 'setup')}, + follow_redirects=False, + ) + assert result.status_code == 303 + assert result.headers['location'] == '/setup/pcloud' + cookie = client.cookies.get(main.COOKIE) + principal = env.store.get('session', hashlib.sha256(cookie.encode()).hexdigest()) + return principal['csrf'] + + +def connection(env, uid=UID): + env.store.put( + 'connection', + uid, + uid, + { + 'access_token': 'synthetic-provider-token', + 'status': 'active', + 'location_id': 1, + 'email': 'synthetic@example.invalid', + 'generation': 'generation-a', + }, + ) + + +def settings(env, **values): + env.store.put( + 'settings', + UID, + UID, + { + 'folder_name': 'Omi Conversations', + 'save_summary': True, + 'save_transcript': False, + 'save_audio': False, + 'consent_version': 1, + **values, + }, + ) + + +def deliver(env, value=None, uid=UID, headers=None): + value = value or payload() + auth = ( + headers + if headers is not None + else TRANSPORT.delivery_headers(uid, 'https://pcloud.example.invalid/conversation', value) + ) + return env.client.post('/conversation', params={'uid': uid}, json=value, headers=auth) + + +def success_provider(env): + calls = [] + env.monkeypatch.setattr(PCloudClient, 'ensure_folder', lambda *a: (123, None)) + + def upload(self, folder_ref, filename, content, overwrite=True): + calls.append((filename, content if isinstance(content, bytes) else content.read())) + return BackupUploadResult('id', filename, '/backup/' + filename, 1, 'now'), None + + env.monkeypatch.setattr(PCloudClient, 'upload_file', upload) + return calls + + +@pytest.mark.parametrize( + 'path,method', + [ + ('/setup/pcloud?uid=known-victim', 'get'), + ('/auth/pcloud', 'post'), + ('/settings', 'post'), + ('/disconnect', 'post'), + ], +) +def test_uid_does_not_authorize(env, path, method): + response = ( + getattr(env.client, method)( + path, + data={'csrf': 'bogus', 'folder_name': 'x'} if method == 'post' else None, + ) + if method == 'post' + else env.client.get(path) + ) + assert response.status_code == 401 + + +def test_setup_grant_is_uid_bound_one_use_and_scrubbed(env): + grant = TRANSPORT.token(UID, 'setup') + assert env.client.get('/setup/pcloud', params={'uid': 'other', 'omi_token': grant}).status_code == 401 + response = env.client.get('/setup/pcloud', params={'uid': UID, 'omi_token': grant}, follow_redirects=False) + assert response.status_code == 303 + assert 'Secure' in response.headers['set-cookie'] and 'HttpOnly' in response.headers['set-cookie'] + assert env.client.get('/setup/pcloud', params={'uid': UID, 'omi_token': grant}).status_code == 401 + assert env.client.get('/setup/pcloud?uid=other').status_code == 403 + + +@pytest.mark.parametrize('scope', ['status', 'audio', 'delivery']) +def test_other_scopes_cannot_open_setup(env, scope): + assert ( + env.client.get( + '/setup/pcloud', + params={'uid': UID, 'omi_token': TRANSPORT.token(UID, scope)}, + ).status_code + == 401 + ) + + +def test_expired_or_wrong_key_setup_denied(env): + expired = TRANSPORT.token(UID, 'setup', ttl=-1) + wrong = PluginTransport(APP, TRANSPORT.base_url, 'other-key-with-at-least-32-bytes!!').token(UID, 'setup') + for token in (expired, wrong): + assert env.client.get('/setup/pcloud', params={'uid': UID, 'omi_token': token}).status_code == 401 + + +def test_csrf_and_account_are_bound_to_session(env): + csrf = login(env) + for wrong in ('bogus', '非ASCII'): + assert env.client.post('/settings', data={'csrf': wrong, 'folder_name': 'x'}).status_code == 403 + result = env.client.post( + '/settings?uid=victim', + data={'csrf': csrf, 'folder_name': 'mine', 'save_summary': 'on'}, + ) + assert result.status_code == 200 + assert env.store.settings(UID)['folder_name'] == 'mine' + assert env.store.settings('victim')['folder_name'] == 'Omi Conversations' + + +def test_setup_status_requires_grant_and_saved_preferences(env): + connection(env) + params = {'uid': UID, 'omi_token': TRANSPORT.token(UID, 'status')} + assert env.client.get('/setup-completed', params=params).json() == {'is_setup_completed': False} + settings(env) + assert env.client.get('/setup-completed', params=params).json() == {'is_setup_completed': True} + assert env.client.get('/setup-completed', params={**params, 'uid': 'other'}).status_code == 401 + + +@pytest.mark.parametrize('region,host', [(1, 'api.pcloud.com'), (2, 'eapi.pcloud.com')]) +def test_oauth_uses_documented_callback_region(env, region, host): + csrf = login(env) + initiation = env.client.post('/auth/pcloud', data={'csrf': csrf}, follow_redirects=False) + state = parse_qs(urlsplit(initiation.headers['location']).query)['state'][0] + response = SimpleNamespace( + status_code=200, + raise_for_status=lambda: None, + json=lambda: {'result': 0, 'access_token': 'synthetic-oauth-token', 'uid': 123}, + ) + calls = [] + env.monkeypatch.setattr(requests, 'post', lambda url, **kw: calls.append(url) or response) + env.monkeypatch.setattr( + PCloudClient, + 'get_user_info', + lambda self: ({'email': 'synthetic@example.invalid'}, None), + ) + result = env.client.get( + '/auth/pcloud/callback', + params={ + 'state': state, + 'code': 'synthetic-code', + 'locationid': region, + 'hostname': host, + }, + ) + assert result.status_code == 200 + assert calls == [f'https://{host}/oauth2_token'] + saved = env.store.get('connection', UID) + assert saved['location_id'] == region and saved['userid'] == 123 + assert env.client.get('/auth/pcloud/callback', params={'state': state, 'code': 'again'}).status_code == 400 + + +def test_oauth_rejects_untrusted_host_and_stale_state(env): + login(env) + env.store.put('oauth', 'expired', UID, {'uid': UID, 'session_csrf': 'x'}, ttl=-1) + assert env.client.get('/auth/pcloud/callback', params={'state': 'expired', 'code': 'x'}).status_code == 400 + csrf = env.store.get( + 'session', + hashlib.sha256(env.client.cookies.get(main.COOKIE).encode()).hexdigest(), + )['csrf'] + env.store.put('oauth', 'host', UID, {'uid': UID, 'session_csrf': csrf}) + assert ( + env.client.get( + '/auth/pcloud/callback', + params={ + 'state': 'host', + 'code': 'x', + 'locationid': 2, + 'hostname': 'metadata.google.internal', + }, + ).status_code + == 400 + ) + + +def begin_oauth(env, csrf, client=None): + client = client or env.client + initiation = client.post('/auth/pcloud', data={'csrf': csrf}, follow_redirects=False) + assert initiation.status_code == 303 + return parse_qs(urlsplit(initiation.headers['location']).query)['state'][0] + + +def finish_oauth(env, state, client=None): + client = client or env.client + return client.get( + '/auth/pcloud/callback', + params={'state': state, 'code': f'code-{state[:6]}', 'locationid': 1, 'hostname': 'api.pcloud.com'}, + ) + + +def synthetic_provider(env, before_exchange=None, revoke_ok=True): + """Synthetic token endpoint: each exchange issues a distinct token; logouts are recorded.""" + issued, revoked = [], [] + + def exchange(url, **kwargs): + if before_exchange: + before_exchange() + issued.append(f'token-{len(issued) + 1}') + return SimpleNamespace( + status_code=200, + raise_for_status=lambda: None, + json=lambda token=issued[-1]: {'result': 0, 'access_token': token, 'uid': 123}, + ) + + env.monkeypatch.setattr(requests, 'post', exchange) + env.monkeypatch.setattr(PCloudClient, 'get_user_info', lambda self: ({'email': 'synthetic@example.invalid'}, None)) + env.monkeypatch.setattr( + PCloudClient, 'logout', lambda self: revoked.append(self.access_token) or ((True, None) if revoke_ok else (False, 'x')) + ) + return issued, revoked + + +def test_disconnect_retires_an_older_pending_authorization(env): + csrf = login(env) + issued, revoked = synthetic_provider(env) + first, second = begin_oauth(env, csrf), begin_oauth(env, csrf) + assert finish_oauth(env, first).status_code == 200 + assert env.client.post('/disconnect', data={'csrf': csrf}).status_code == 200 + late = finish_oauth(env, second) + assert late.status_code == 409 and 'disconnected' in late.text.lower() + assert env.store.get('connection', UID) is None + assert issued == ['token-1'] and revoked == ['token-1'] # the retired code is never exchanged + + +def test_disconnect_without_a_connection_still_retires_pending_authorization(env): + csrf = login(env) + issued, _ = synthetic_provider(env) + pending = begin_oauth(env, csrf) + assert env.client.post('/disconnect', data={'csrf': csrf}).status_code == 200 + assert finish_oauth(env, pending).status_code == 409 + assert env.store.get('connection', UID) is None and issued == [] + # A genuinely new Connect after Disconnect works. + assert finish_oauth(env, begin_oauth(env, csrf)).status_code == 200 + assert env.store.get('connection', UID)['access_token'] == 'token-1' + + +@pytest.mark.parametrize('revoke_ok', [True, False]) +def test_disconnect_during_token_exchange_cannot_revive_the_account(env, revoke_ok): + csrf = login(env) + exchanging, release = concurrent.futures.Future(), concurrent.futures.Future() + + def pause(): + if not exchanging.done(): + exchanging.set_result(True) + release.result(timeout=10) + + issued, revoked = synthetic_provider(env, before_exchange=pause, revoke_ok=revoke_ok) + state = begin_oauth(env, csrf) + with concurrent.futures.ThreadPoolExecutor(1) as pool: + pending = pool.submit(finish_oauth, env, state) + exchanging.result(timeout=10) + assert env.client.post('/disconnect', data={'csrf': csrf}).status_code == 200 + release.set_result(True) + result = pending.result(timeout=10) + assert result.status_code == 409 and 'disconnected' in result.text.lower() + assert env.store.get('connection', UID) is None + assert revoked == issued == ['token-1'] # the stale grant is revoked, not stored + assert ('not confirmed' in result.text) is not revoke_ok + + +def test_stale_exchange_never_overwrites_a_newer_connection(env): + csrf = login(env) + exchanging, release = concurrent.futures.Future(), concurrent.futures.Future() + + def pause(): + if not exchanging.done(): + exchanging.set_result(True) + release.result(timeout=10) + + issued, revoked = synthetic_provider(env, before_exchange=pause) + stale = begin_oauth(env, csrf) + with concurrent.futures.ThreadPoolExecutor(1) as pool: + pending = pool.submit(finish_oauth, env, stale) + exchanging.result(timeout=10) + assert env.client.post('/disconnect', data={'csrf': csrf}).status_code == 200 + assert finish_oauth(env, begin_oauth(env, csrf)).status_code == 200 # new attempt issues token-1 + release.set_result(True) + assert pending.result(timeout=10).status_code == 409 # the stale attempt then issues token-2 + assert env.store.get('connection', UID)['access_token'] == 'token-1' + assert revoked == ['token-2'] + + +def test_disconnect_leaves_other_accounts_and_canceled_jobs_alone(env): + csrf = login(env) + other = TestClient(main.app, base_url='https://pcloud.example.invalid') + other_csrf = login(env, uid='synthetic-user-b', client=other) + synthetic_provider(env) + other_state = begin_oauth(env, other_csrf, client=other) + connection(env) + settings(env) + assert deliver(env).status_code == 202 + assert env.client.post('/disconnect', data={'csrf': csrf}).status_code == 200 + assert finish_oauth(env, other_state, client=other).status_code == 200 + assert env.store.get('connection', 'synthetic-user-b')['status'] == 'active' + assert finish_oauth(env, begin_oauth(env, csrf)).status_code == 200 + worker.run_next_job(env.store) + assert [job['state'] for job in env.store.jobs(UID)] == ['canceled'] + other.close() + + +def test_failed_revocation_stops_exports_but_preserves_recovery_token(env): + csrf = login(env) + connection(env) + deliver(env) + env.monkeypatch.setattr(PCloudClient, 'logout', lambda self: (False, 'provider secret error')) + result = env.client.post('/disconnect', data={'csrf': csrf}) + assert result.status_code == 502 and 'not confirmed' in result.text + assert 'provider secret' not in result.text + assert env.store.get('connection', UID)['status'] == 'revoke_pending' + assert env.store.jobs(UID)[0]['state'] == 'canceled' + assert 'files' not in env.store.jobs(UID)[0] + assert deliver(env).status_code == 409 + env.monkeypatch.setattr(PCloudClient, 'logout', lambda self: (True, None)) + assert env.client.post('/disconnect', data={'csrf': csrf}).status_code == 200 + assert env.store.get('connection', UID) is None + + +def test_disconnect_preserves_completed_receipts(env): + csrf = login(env) + connection(env) + deliver(env) + success_provider(env) + worker.run_next_job(env.store) + before = env.store.jobs(UID)[0] + assert before['state'] == 'completed' and before['uploaded'] == ['summary.md'] + env.monkeypatch.setattr(PCloudClient, 'logout', lambda self: (False, 'synthetic outage')) + result = env.client.post('/disconnect', data={'csrf': csrf}) + after = env.store.jobs(UID)[0] + assert after == before + assert 'Completed' in result.text and '1 file saved' in result.text + + +def test_disconnect_scrubs_partial_payload_but_keeps_saved_file_receipts(env): + csrf = login(env) + connection(env) + settings(env, save_transcript=True) + deliver(env) + identity, uid, lease, job = env.store.claim() + job['uploaded'] = ['summary.md'] + env.store.settle(identity, uid, lease, job, 'failed') + env.monkeypatch.setattr(PCloudClient, 'logout', lambda self: (True, None)) + env.client.post('/disconnect', data={'csrf': csrf}) + after = env.store.jobs(UID)[0] + assert after['state'] == 'canceled' and after['uploaded'] == ['summary.md'] + assert after['title'] == 'Same title' + assert not {'files', 'audio_token', 'authorization_token'} & after.keys() + + +@pytest.mark.parametrize('tamper', ['body', 'uid', 'scope', 'none']) +def test_delivery_signature_covers_identity_and_content(env, tamper): + connection(env) + value = payload() + headers = TRANSPORT.delivery_headers(UID, 'https://pcloud.example.invalid/conversation', value) + uid = UID + if tamper == 'body': + value['structured']['overview'] = 'tampered' + elif tamper == 'uid': + uid = 'other' + elif tamper == 'scope': + headers = {'X-Omi-Plugin-Authorization': TRANSPORT.token(UID, 'setup')} + else: + headers = {} + assert deliver(env, value, uid=uid, headers=headers).status_code == 401 + assert env.store.jobs(UID) == [] + + +def test_legacy_missing_consent_does_not_export_or_store_sensitive_content(env): + connection(env) + env.store.put('settings', UID, UID, {'save_transcript': True, 'save_audio': True}) + response = deliver(env) + assert response.status_code == 202 + job = env.store.jobs(UID)[0] + assert list(job['files']) == ['summary.md'] and job['audio_token'] is None + assert 'Synthetic private transcript' not in json.dumps(job) + calls = success_provider(env) + assert worker.run_next_job(env.store) + assert [name for name, _ in calls] == ['summary.md'] + assert env.store.jobs(UID)[0]['state'] == 'completed' + + +def test_discard_and_uid_audio_never_create_cross_conversation_buffers(env): + connection(env) + value = payload() + value['discarded'] = True + assert deliver(env, value).json()['status'] == 'skipped' + assert env.client.post('/audio?uid=' + UID, content=b'private').status_code == 410 + assert env.store.jobs(UID) == [] + + +def test_distinct_conversation_identity_is_unique_but_retries_are_not(env): + connection(env) + a = deliver(env).json()['backup_id'] + again = deliver(env, payload(title='Changed title')).json()['backup_id'] + b = deliver(env, payload('conversation-b')).json()['backup_id'] + assert a == again and a != b + jobs = env.store.jobs(UID) + assert len(jobs) == 2 and jobs[0]['path'] != jobs[1]['path'] + + +def test_failed_partial_upload_preserves_payload_and_resumes_only_missing_files(env): + connection(env) + settings(env, save_transcript=True) + job_id = deliver(env).json()['backup_id'] + env.monkeypatch.setattr(PCloudClient, 'ensure_folder', lambda *a: (1, None)) + calls = [] + + def first(self, folder_ref, filename, content, overwrite=True): + calls.append(filename) + return ( + (BackupUploadResult('1', filename, '/file', 1, 'now'), None) + if filename == 'summary.md' + else (None, 'secret provider text') + ) + + env.monkeypatch.setattr(PCloudClient, 'upload_file', first) + worker.run_next_job(env.store) + job = env.store.jobs(UID)[0] + assert job['state'] == 'pending' and job['uploaded'] == ['summary.md'] + assert 'transcript.md' in job['files'] and 'secret provider text' not in job['error'] + calls = success_provider(env) + assert env.store.retry(UID, job_id) + worker.run_next_job(Store(str(env.store.path), [env.key])) + assert [name for name, _ in calls] == ['transcript.md'] + done = env.store.jobs(UID)[0] + assert done['state'] == 'completed' and done['files'] == {} and done['audio_token'] is None + + +def test_unfinished_backup_follows_a_destination_folder_change(env): + connection(env) + settings(env, save_transcript=True) + job_id = deliver(env).json()['backup_id'] + folders = [] + env.monkeypatch.setattr(PCloudClient, 'ensure_folder', lambda self, path: (folders.append(path) or 1, None)) + + def partial(self, folder_ref, filename, content, overwrite=True): + if filename == 'summary.md': + return BackupUploadResult('1', filename, '/file', 1, 'now'), None + return None, 'provider failure' + + env.monkeypatch.setattr(PCloudClient, 'upload_file', partial) + worker.run_next_job(env.store) + assert env.store.jobs(UID)[0]['uploaded'] == ['summary.md'] + assert folders and all(path == f'/Omi Conversations/{job_id}' for path in folders) + + settings(env, save_transcript=True, folder_name='Work Notes') + folders.clear() + calls = success_provider(env) + env.monkeypatch.setattr(PCloudClient, 'ensure_folder', lambda self, path: (folders.append(path) or 1, None)) + assert env.store.retry(UID, job_id) + worker.run_next_job(env.store) + + assert folders and all(path == f'/Work Notes/{job_id}' for path in folders) + assert sorted(name for name, _ in calls) == ['summary.md', 'transcript.md'], 'the new folder gets the whole backup' + assert env.store.jobs(UID)[0]['state'] == 'completed' + + +def test_completed_backup_receipts_survive_expiry(env): + connection(env) + job_id = deliver(env).json()['backup_id'] + success_provider(env) + worker.run_next_job(env.store) + assert env.store.jobs(UID)[0]['uploaded'] == ['summary.md'] + with env.store.transaction() as conn: + conn.execute('UPDATE jobs SET expires=? WHERE id=?', (time.time() - 1, job_id)) + assert env.store.claim() is None # runs the expiry scrub + done = env.store.jobs(UID)[0] + assert done['state'] == 'completed' and done['uploaded'] == ['summary.md'] + + +def test_failed_backups_stay_reachable_beyond_the_recent_history(env): + failed = env.store.enqueue(UID, 'old-failed', {'generation': 'g', 'files': {}, 'audio_token': None}) + with env.store.transaction() as conn: + conn.execute('UPDATE jobs SET state=? WHERE id=?', ('failed', failed)) + for index in range(35): + newer = env.store.enqueue(UID, f'newer-{index}', {'generation': 'g', 'files': {}, 'audio_token': None}) + with env.store.transaction() as conn: + conn.execute('UPDATE jobs SET state=? WHERE id=?', ('completed', newer)) + listed = env.store.jobs(UID) + assert failed in [job['id'] for job in listed], 'an old failed backup must stay retryable from settings' + assert len([job for job in listed if job['state'] == 'completed']) <= 30 + + +def test_opt_out_and_relink_fence_queued_jobs(env): + connection(env) + settings(env, save_transcript=True) + deliver(env) + settings(env, save_transcript=False) + calls = success_provider(env) + worker.run_next_job(env.store) + assert [name for name, _ in calls] == ['summary.md'] + deliver(env, payload('new-conversation')) + saved = env.store.get('connection', UID) + saved['generation'] = 'different-account-generation' + env.store.put('connection', UID, UID, saved) + worker.run_next_job(env.store) + assert env.store.jobs(UID)[0]['state'] == 'canceled' + + +def test_store_claim_nonce_rotation_and_encryption_are_durable(env): + connection(env) + job_id = deliver(env).json()['backup_id'] + assert env.store.consume_nonce('nonce', UID, 900) + assert not env.store.consume_nonce('nonce', UID, 900) + claimed = env.store.claim() + assert claimed[0] == job_id and env.store.claim() is None + raw = env.store.path.read_bytes() + assert b'synthetic-provider-token' not in raw and b'Synthetic summary' not in raw + new_key = Fernet.generate_key() + env.store.cipher = MultiFernet([Fernet(new_key), Fernet(env.key.encode())]) + env.store.rotate_keys() + env.store.cipher = MultiFernet([Fernet(new_key)]) + assert env.store.get('connection', UID)['status'] == 'active' + assert env.store.jobs(UID)[0]['id'] == job_id + + +def test_job_expiry_scrubs_pending_content(env): + connection(env) + deliver(env) + with env.store.transaction() as conn: + conn.execute('UPDATE jobs SET expires=?', (time.time() - 1,)) + assert env.store.claim() is None + job = env.store.jobs(UID)[0] + assert job['state'] == 'expired' and 'files' not in job and 'audio_token' not in job + + +def test_crashed_lease_reclaims_but_cannot_commit_stale_work(env): + connection(env) + deliver(env) + first = env.store.claim() + with env.store.transaction() as conn: + conn.execute('UPDATE jobs SET due=0') + reopened = Store(str(env.store.path), [env.key]) + second = reopened.claim() + assert first[0] == second[0] and first[2] != second[2] + with pytest.raises(RuntimeError): + env.store.settle(first[0], UID, first[2], first[3], 'completed') + + +def test_duplicate_admission_is_atomic_across_store_instances(env): + job = {'generation': 'a', 'files': {}, 'audio_token': None} + + def enqueue(index): + store = Store(str(env.store.path), [env.key]) + return store.enqueue(UID, 'same-conversation', dict(job)) + + with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool: + identities = list(pool.map(enqueue, range(8))) + assert len(set(identities)) == 1 and len(env.store.jobs(UID)) == 1 + + +def test_fresh_delivery_replaces_an_expired_or_canceled_backup(env): + first = env.store.enqueue(UID, 'conv-r', {'generation': 'old', 'files': {}, 'audio_token': None}) + with env.store.transaction() as conn: + conn.execute('UPDATE jobs SET expires=? WHERE id=?', (time.time() - 1, first)) + assert env.store.claim() is None + assert env.store.job_state(UID, first) == 'expired' + + again = env.store.enqueue(UID, 'conv-r', {'generation': 'new', 'files': {}, 'audio_token': None}) + assert again == first and env.store.job_state(UID, first) == 'pending' + claimed = env.store.claim() + assert claimed[0] == first and claimed[3]['generation'] == 'new' + + env.store.cancel_jobs(UID) + assert env.store.job_state(UID, first) == 'canceled' + env.store.enqueue(UID, 'conv-r', {'generation': 'reconnected', 'files': {}, 'audio_token': None}) + assert env.store.job_state(UID, first) == 'pending' + + +def test_completed_and_active_backups_still_deduplicate(env): + identity = env.store.enqueue(UID, 'conv-d', {'generation': 'one', 'files': {}, 'audio_token': None}) + env.store.enqueue(UID, 'conv-d', {'generation': 'two', 'files': {}, 'audio_token': None}) + claimed = env.store.claim() + assert claimed[3]['generation'] == 'one', 'a pending job is not replaced' + env.store.settle(identity, UID, claimed[2], {**claimed[3], 'uploaded': ['summary.md']}, 'completed') + env.store.enqueue(UID, 'conv-d', {'generation': 'three', 'files': {}, 'audio_token': None}) + assert env.store.job_state(UID, identity) == 'completed', 'a completed backup is never re-uploaded' + + +def test_backup_summary_follows_the_shared_primary_summary_vectors(): + from presentation import select_primary_summary + + vectors = json.loads( + (Path(__file__).resolve().parents[2] / 'contracts' / 'parity' / 'conversation_summary.json').read_text() + ) + assert vectors['cases'] + for case in vectors['cases']: + assert select_primary_summary(case['conversation']) == case['expected'], case['id'] + + +def test_backup_summary_uses_sections_when_the_overview_is_empty(): + from omi_plugin_sdk.models import Conversation + from presentation import summary_markdown + + value = payload() + value['structured']['overview'] = '' + value['structured']['sections'] = [{'heading': 'Decisions', 'body_markdown': '- Ship the backup fix.'}] + body = summary_markdown(Conversation.model_validate(value)) + assert '## Decisions\n\n- Ship the backup fix.' in body + + value['apps_results'] = [{'app_id': 'notes-app', 'content': 'App-written summary'}] + assert 'App-written summary' in summary_markdown(Conversation.model_validate(value)) + + +def test_expired_state_nonce_cannot_be_consumed(env): + env.store.put('oauth', 'old', UID, {'uid': UID}, ttl=-1) + assert env.store.get('oauth', 'old', consume=True) is None + + +def test_wrong_encryption_key_fails_readiness(env): + env.store.check_ready() + wrong = Store(str(env.store.path), [Fernet.generate_key().decode()]) + with pytest.raises(InvalidToken): + wrong.check_ready() + + +def test_corrupt_job_is_visible_and_does_not_starve_other_jobs(env): + connection(env) + identity = deliver(env).json()['backup_id'] + deliver(env, payload('healthy-conversation')) + with env.store.transaction() as conn: + conn.execute('UPDATE jobs SET blob=? WHERE id=?', (b'corrupt-ciphertext', identity)) + assert env.store.claim() is None + job = next(item for item in env.store.jobs(UID) if item['id'] == identity) + assert job['state'] == 'unreadable' + success_provider(env) + assert worker.run_next_job(env.store) + assert env.store.jobs(UID)[0]['state'] == 'completed' + + +@pytest.mark.parametrize('status', [403, 404, 503]) +def test_queued_copy_stops_if_omi_access_is_no_longer_confirmed(env, status): + connection(env) + deliver(env) + calls = success_provider(env) + env.monkeypatch.setattr(requests, 'head', lambda *args, **kwargs: SimpleNamespace(status_code=status)) + worker.run_next_job(env.store) + assert calls == [] and env.store.jobs(UID)[0]['state'] == 'pending' + + +def test_terminal_retry_budget_is_visible_and_can_be_retried_by_owner(env): + connection(env) + identity = deliver(env).json()['backup_id'] + env.monkeypatch.setattr(PCloudClient, 'ensure_folder', lambda *args: (None, 'provider error')) + for index in range(5): + with env.store.transaction() as conn: + conn.execute('UPDATE jobs SET due=0') + worker.run_next_job(env.store) + assert env.store.jobs(UID)[0]['state'] == 'failed' + csrf = login(env) + assert env.client.post(f'/backups/{identity}/retry', data={'csrf': csrf}).status_code == 200 + assert env.store.jobs(UID)[0]['state'] == 'pending' + assert env.store.retry('other-user', identity) is False + + +def test_completed_replay_is_not_reported_as_newly_queued(env): + connection(env) + deliver(env) + success_provider(env) + worker.run_next_job(env.store) + result = deliver(env) + assert result.status_code == 200 and result.json()['status'] == 'completed' + + +def test_setup_escapes_private_text_and_matches_saved_consent(env): + connection(env) + csrf = login(env) + result = env.client.post( + '/settings', + data={ + 'csrf': csrf, + 'folder_name': 'Folder ', + 'save_summary': 'on', + }, + ) + assert result.status_code == 200 + assert '