diff --git a/mobile/lib/shared/mentions/agent_authorization.dart b/mobile/lib/shared/mentions/agent_authorization.dart new file mode 100644 index 00000000000..1625db85803 --- /dev/null +++ b/mobile/lib/shared/mentions/agent_authorization.dart @@ -0,0 +1,162 @@ +part of 'agent_identity_provider.dart'; + +/// Fresh exact-key policy and relay-signed membership. Directory entries are +/// hints; only this read may authorize agent mentions at a destination. +Future> readAgentAuthorization( + RelaySessionNotifier session, + Set requestedKeys, { + required String? viewer, + String? channelId, + required bool Function() isCurrent, +}) async { + void check() { + if (!isCurrent()) throw StateError('Agent authorization scope changed'); + } + + check(); + if (requestedKeys.isEmpty) return []; + if (viewer == null || + requestedKeys.length > 1000 || + requestedKeys.any((key) => !RegExp(r'^[0-9a-f]{64}$').hasMatch(key))) { + throw StateError('Invalid agent authorization request'); + } + final authority = await session.fetchRelaySelf(); + check(); + var membership = await _membershipPages( + session, + authority, + viewer, + channelId, + check, + ); + if (channelId == null) { + // #p discovers destinations, not authority: a replacement removing the + // viewer cannot match that filter. Re-read each exact coordinate without + // #p, and never retain a discovery snapshot if its head is absent. + final destinations = membership + .where((e) => e.kind == 39002 && e.pubkey == authority) + .map((e) => e.getTagValue('d')) + .whereType() + .toSet(); + if (destinations.length > 1000) { + throw StateError('Membership destination budget exceeded'); + } + membership = await _queryAgentFilters(session, [ + for (final destination in destinations) + NostrFilter( + kinds: const [39002], + authors: [authority], + tags: { + '#d': [destination], + }, + limit: 1, + ), + ], checkCurrent: check); + } + check(); + final runtime = await _queryAgentFilters(session, [ + for (final key in requestedKeys) + NostrFilter(kinds: const [10100], authors: [key], limit: 1), + ], checkCurrent: check); + check(); + // Guard every query stage, not merely the result: a mutable session must not + // continue an old account/community request against the newly selected relay. + final agents = await resolveAgentPolicies( + session, + runtime.where((e) => requestedKeys.contains(e.pubkey)).toList(), + requestedKeys: requestedKeys, + checkCurrent: check, + ); + check(); + final latest = {}; + for (final event in membership) { + if (event.kind != 39002 || event.pubkey != authority) { + continue; + } + final destination = event.getTagValue('d'); + if (destination == null || + (channelId != null && destination != channelId)) { + continue; + } + final previous = latest[destination]; + if (previous == null || _newer(event, previous)) { + latest[destination] = event; + } + } + final result = []; + for (final agent in agents) { + final owned = agent.ownerPubkey == viewer; + final channels = []; + for (final entry in latest.entries) { + if (!verifySignedEvent(entry.value) || + entry.value.tags + .where((tag) => tag.isNotEmpty && tag[0] == 'd') + .length != + 1) { + continue; + } + final people = entry.value.tags.where( + (tag) => tag.length >= 2 && tag[0] == 'p', + ); + if (!people.any((tag) => tag[1] == viewer)) continue; + if (people.any( + (tag) => + tag[1] == agent.pubkey && + (owned || (tag.length >= 4 && tag[3] == 'bot')), + )) { + channels.add(entry.key); + } + } + if (!owned && channels.isEmpty) continue; + result.add( + AgentDirectoryEntry( + pubkey: agent.pubkey, + ownerPubkey: agent.ownerPubkey, + displayName: agent.displayName, + respondTo: agent.respondTo, + respondToAllowlist: agent.respondToAllowlist, + channelIds: channels, + ), + ); + } + return result; +} + +Future> _membershipPages( + RelaySessionNotifier session, + String authority, + String viewer, + String? channelId, + void Function() check, +) async { + final result = []; + NostrEvent? cursor; + for (var page = 0; page < 200; page++) { + check(); + final events = await session.queryRelay([ + NostrFilter( + kinds: const [39002], + authors: [authority], + tags: { + if (channelId == null) '#p': [viewer], + if (channelId != null) '#d': [channelId], + }, + limit: 500, + until: cursor?.createdAt, + extensions: {if (cursor != null) 'before_id': cursor.id}, + ), + ]); + check(); + result.addAll(events); + if (events.length < 500) return result; + final next = events.last; + if (cursor != null && + (next.createdAt > cursor.createdAt || + (next.createdAt == cursor.createdAt && + next.id.compareTo(cursor.id) <= 0))) { + throw StateError('Membership pagination did not advance'); + } + cursor = next; + } + throw StateError('Membership query exceeded its page budget'); +} diff --git a/mobile/lib/shared/mentions/agent_identity_provider.dart b/mobile/lib/shared/mentions/agent_identity_provider.dart index e5be060be40..f8b4dc6cf49 100644 --- a/mobile/lib/shared/mentions/agent_identity_provider.dart +++ b/mobile/lib/shared/mentions/agent_identity_provider.dart @@ -9,6 +9,7 @@ import '../../shared/crypto/signed_event.dart'; import '../../shared/relay/relay.dart'; part 'agent_policy.dart'; +part 'agent_authorization.dart'; /// A relay agent parsed from its kind:10100 agent-profile event. /// @@ -18,6 +19,9 @@ part 'agent_policy.dart'; class AgentDirectoryEntry { final String pubkey; final String? displayName; + + /// Owner authenticated by the agent's verified NIP-OA kind:0 attestation, + /// not a runtime ownership claim. Null means no authenticated owner here. final String? ownerPubkey; final String? respondTo; final List respondToAllowlist; @@ -70,8 +74,19 @@ final agentDirectoryProvider = FutureProvider>(( final sessionState = ref.watch(relaySessionProvider); if (sessionState.status != SessionStatus.connected) return const []; final session = ref.read(relaySessionProvider.notifier); + final config = ref.read(relayConfigProvider); + var disposed = false; + ref.onDispose(() => disposed = true); + final viewer = ref.read(myPubkeyProvider); final events = await session.fetchHistory(NostrFilters.agentProfiles()); - return resolveAgentPolicies(session, events); + if (disposed) return []; + return readAgentAuthorization( + session, + events.map((event) => event.pubkey).toSet(), + viewer: viewer, + isCurrent: () => + !disposed && identical(config, ref.read(relayConfigProvider)), + ); }); /// Verified NIP-OA owner pubkey per agent pubkey, from the agents' kind:0 diff --git a/mobile/lib/shared/mentions/agent_policy.dart b/mobile/lib/shared/mentions/agent_policy.dart index be7f6642b88..9d2d57a2742 100644 --- a/mobile/lib/shared/mentions/agent_policy.dart +++ b/mobile/lib/shared/mentions/agent_policy.dart @@ -9,10 +9,12 @@ bool _newer(NostrEvent event, NostrEvent previous) => /// is distinct from a failed read; failures must never revive runtime access. Future> _queryAgentFilters( RelaySessionNotifier session, - List filters, -) async { + List filters, { + void Function()? checkCurrent, +}) async { final events = []; for (var start = 0; start < filters.length; start += 10) { + checkCurrent?.call(); events.addAll( await session.queryRelay(filters.skip(start).take(10).toList()), ); @@ -24,8 +26,10 @@ Future> _queryAgentFilters( /// directory. This does not expand discovery to owner-only coordinates yet. Future> resolveAgentPolicies( RelaySessionNotifier session, - List runtimeEvents, -) async { + List runtimeEvents, { + Set? requestedKeys, + void Function()? checkCurrent, +}) async { final latest = {}; for (final event in runtimeEvents.where((event) => event.kind == 10100)) { final previous = latest[event.pubkey]; @@ -33,19 +37,21 @@ Future> resolveAgentPolicies( latest[event.pubkey] = event; } } + final keys = requestedKeys ?? latest.keys.toSet(); final profiles = latestProfileEvents( await _queryAgentFilters(session, [ - for (final key in latest.keys) + for (final key in keys) NostrFilter(kinds: const [0], authors: [key], limit: 1), - ]), + ], checkCurrent: checkCurrent), ); final owners = {}; for (final profile in profiles.values) { final owner = verifiedOaOwnerPubkey(profile); - if (owner != null && latest.containsKey(profile.pubkey)) { + if (owner != null && keys.contains(profile.pubkey)) { owners[profile.pubkey] = owner; } } + checkCurrent?.call(); final policies = await _queryAgentFilters(session, [ for (final owner in owners.entries) NostrFilter( @@ -56,7 +62,8 @@ Future> resolveAgentPolicies( }, limit: 1, ), - ]); + ], checkCurrent: checkCurrent); + checkCurrent?.call(); return mergeAgentPolicies(latest.values, policies, owners); } diff --git a/mobile/lib/shared/relay/relay_http_query_client.dart b/mobile/lib/shared/relay/relay_http_query_client.dart index 615b28f64cf..73ea4d15884 100644 --- a/mobile/lib/shared/relay/relay_http_query_client.dart +++ b/mobile/lib/shared/relay/relay_http_query_client.dart @@ -15,6 +15,29 @@ class RelayHttpQueryClient { _ClientGeneration? _currentGeneration; final Set<_ClientGeneration> _generations = {}; + /// Read relay metadata over the same owned transport as query requests. + Future get( + Uri url, { + required Map headers, + required Duration timeout, + }) async { + final generation = _injectedClient == null + ? (_currentGeneration ??= _createGeneration()) + : null; + generation?.acquire(); + try { + return await (_injectedClient ?? generation!.client) + .get(url, headers: headers) + .timeout(timeout); + } on TimeoutException { + if (identical(_currentGeneration, generation)) _currentGeneration = null; + generation?.retire(); + rethrow; + } finally { + generation?.release(); + } + } + Future post( Uri url, { required Map headers, diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index 6a787cce129..b761f23c662 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -149,6 +149,25 @@ class RelaySessionNotifier extends Notifier { return const SessionState(status: SessionStatus.disconnected); } + /// NIP-11 `self` is the membership signer; `pubkey` is only contact metadata. + Future fetchRelaySelf() async { + final config = ref.read(relayConfigProvider); + final response = await _httpQueryClient.get( + Uri.parse(config.baseUrl), + headers: const {'Accept': 'application/nostr+json'}, + timeout: const Duration(seconds: 8), + ); + if (response.statusCode != 200) { + throw RelayException(response.statusCode, 'Relay authority unavailable'); + } + final data = jsonDecode(response.body); + final key = data is Map ? data['self'] : null; + if (key is! String || !RegExp(r'^[0-9a-fA-F]{64}$').hasMatch(key)) { + throw const FormatException('Relay membership authority unavailable'); + } + return key.toLowerCase(); + } + /// Execute a one-shot query via the relay's HTTP bridge (`POST /query`). Future> queryRelay( List filters, { diff --git a/mobile/test/shared/mentions/agent_authorization_test.dart b/mobile/test/shared/mentions/agent_authorization_test.dart new file mode 100644 index 00000000000..2d5e4bbc270 --- /dev/null +++ b/mobile/test/shared/mentions/agent_authorization_test.dart @@ -0,0 +1,316 @@ +import 'dart:convert'; + +import 'package:buzz/shared/mentions/agent_identity_provider.dart'; +import 'package:buzz/features/channels/mentions/mention_candidates.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart'; +import 'package:nostr/nostr.dart' as nostr; + +import '../crypto/nip_oa_test.dart' show authTag, profile; +import 'agent_policy_test.dart' show PolicySession, signed; + +class _ReadSession extends PolicySession { + _ReadSession(super.events, this.authority); + final String authority; + void Function(List)? onQuery; + List Function(List)? page; + @override + Future fetchRelaySelf() async => authority; + @override + Future> queryRelay( + List filters, { + Duration timeout = const Duration(seconds: 8), + }) async { + onQuery?.call(filters); + if (page != null && filters.first.kinds.contains(39002)) { + return page!(filters); + } + return super.queryRelay(filters); + } +} + +void main() { + final owner = nostr.Keys.generate(); + final agent = nostr.Keys.generate(); + final relay = nostr.Keys.generate(); + final other = nostr.Keys.generate(); + final owned = profile(agent, [authTag(owner, agent.public)]); + final policy = signed( + owner, + 30177, + {'name': 'Agent', 'parallelism': 1, 'respond_to': 'anyone'}, + tags: [ + ['d', agent.public], + ], + ); + final runtime = signed(agent, 10100, { + 'name': 'Agent', + 'respond_to': 'anyone', + 'channel_ids': ['forged-channel'], + }); + NostrEvent members({ + String channel = 'room', + nostr.Keys? signer, + int time = 100, + bool includeAgent = true, + bool includeViewer = true, + }) => signed( + signer ?? relay, + 39002, + '', + time: time, + tags: [ + ['d', channel], + if (includeViewer) ['p', owner.public], + if (includeAgent) ['p', agent.public, '', 'member'], + ], + ); + + Future> read( + _ReadSession session, { + bool Function()? isCurrent, + String? destination = 'room', + }) async { + final container = ProviderContainer( + overrides: [relaySessionProvider.overrideWith(() => session)], + ); + container.read(relaySessionProvider); + try { + return await readAgentAuthorization( + session, + {agent.public}, + viewer: owner.public, + channelId: destination, + isCurrent: isCurrent ?? () => true, + ); + } finally { + container.dispose(); + } + } + + test( + 'owned agent needs no runtime and ordinary membership role is sufficient', + () async { + final session = _ReadSession([owned, policy, members()], relay.public); + final result = await read(session); + expect(result.single.channelIds, ['room']); + final filter = session.queries.first; + expect(filter.authors, [relay.public]); + expect(filter.tags, { + '#d': ['room'], + }); + expect( + session.queries.where((q) => q.kinds.contains(10100)).single.authors, + [agent.public], + ); + }, + ); + + test( + 'runtime cannot assert membership; exact destination excludes other rooms', + () async { + final result = await read( + _ReadSession([ + owned, + policy, + runtime, + members(channel: 'elsewhere'), + ], relay.public), + ); + expect(result.single.channelIds, isEmpty); + }, + ); + + test( + 'only relay self signature is authority; latest removals do not fall back', + () async { + for (final snapshots in [ + [members(signer: other)], + [ + NostrEvent.fromJson({...members().toJson(), 'content': 'tampered'}), + ], + [members(), members(time: 101, includeAgent: false)], + [ + members(), + NostrEvent.fromJson({ + ...members(time: 101).toJson(), + 'content': 'tampered', + }), + ], + [members(), members(time: 101, includeViewer: false)], + ]) { + final result = await read( + _ReadSession([owned, policy, ...snapshots], relay.public), + ); + expect(result.single.channelIds, isEmpty); + } + }, + ); + + test( + 'global discovery re-reads removal heads without viewer filter', + () async { + final old = signed( + relay, + 39002, + '', + tags: [ + ['d', 'room'], + ['p', owner.public], + ['p', agent.public, '', 'bot'], + ], + ); + final removed = members(time: 101, includeViewer: false); + final session = _ReadSession([runtime], relay.public); + var removedNow = false; + session.page = (filters) { + final filter = filters.single; + expect(filter.authors, [relay.public]); + if (filter.tags.containsKey('#p')) { + expect(filter.tags['#p'], [owner.public]); + return [old]; // The removed head cannot match #p=viewer. + } + expect(filter.tags, { + '#d': ['room'], + }); + expect(filter.limit, 1); + return [removedNow ? removed : old]; + }; + final before = await read(session, destination: null); + expect(before.single.channelIds, ['room']); + removedNow = true; + final after = await read( + _ReadSession([runtime], relay.public)..page = session.page, + destination: null, + ); + expect(after, isEmpty); + expect( + buildMentionCandidates( + members: [], + relayAgents: after, + sharedChannelIds: {'room'}, + userCache: {}, + ownerByAgentPubkey: {}, + currentPubkey: owner.public, + ), + isEmpty, + ); + }, + ); + + test('same-second membership tie is deterministic', () async { + final yes = members(); + final no = members(includeAgent: false); + for (final snapshots in [ + [yes, no], + [no, yes], + ]) { + final result = await read( + _ReadSession([owned, policy, ...snapshots], relay.public), + ); + expect(result.single.channelIds.isNotEmpty, yes.id.compareTo(no.id) < 0); + } + }); + + test('scope changes stop subsequent policy queries and results', () async { + var current = true; + final session = _ReadSession([owned, policy, members()], relay.public); + session.onQuery = (filters) { + if (filters.any((f) => f.kinds.contains(0))) current = false; + }; + await expectLater( + read(session, isCurrent: () => current), + throwsStateError, + ); + expect(session.queries.any((q) => q.kinds.contains(30177)), isFalse); + }); + + test( + 'membership failure propagates, not an authoritative empty result', + () async { + final session = _ReadSession([owned, policy], relay.public); + session.onQuery = (filters) { + if (filters.first.kinds.contains(39002)) { + throw StateError('query unavailable'); + } + }; + await expectLater(read(session), throwsStateError); + }, + ); + + test( + 'membership pagination advances equal-time cursor and rejects stalled pages', + () async { + final first = members(channel: 'a'); + final last = members(channel: 'z'); + final session = _ReadSession([owned, policy], relay.public); + var calls = 0; + session.page = (filters) { + if (filters.first.tags.containsKey('#d')) return [first, last]; + calls++; + if (calls == 1) return List.filled(500, first); + expect(filters.single.until, first.createdAt); + expect(filters.single.extensions['before_id'], first.id); + return [last]; + }; + final result = await read(session, destination: null); + expect(result.single.channelIds.toSet(), {'a', 'z'}); + calls = 0; + final stalled = _ReadSession([owned, policy], relay.public) + ..page = (_) => List.filled(500, first); + await expectLater(read(stalled), throwsStateError); + }, + ); + + test( + 'NIP11 uses self, preserves host and propagates unavailable authority', + () async { + var body = jsonEncode({ + 'self': relay.public.toUpperCase(), + 'pubkey': other.public, + }); + var status = 200; + final client = MockClient((request) async { + expect(request.url.toString(), 'https://tenant.example/'); + expect(request.headers['Accept'], 'application/nostr+json'); + return http.Response(body, status); + }); + final container = ProviderContainer( + overrides: [ + relayConfigProvider.overrideWith(_Config.new), + relaySessionProvider.overrideWith( + () => RelaySessionNotifier(httpClient: client), + ), + ], + ); + try { + final session = container.read(relaySessionProvider.notifier); + expect(await session.fetchRelaySelf(), relay.public); + for (final invalid in [ + jsonEncode({'pubkey': relay.public}), + '{}', + 'not json', + '{"self":"bad"}', + ]) { + body = invalid; + await expectLater(session.fetchRelaySelf(), throwsFormatException); + } + status = 503; + await expectLater( + session.fetchRelaySelf(), + throwsA(isA()), + ); + } finally { + container.dispose(); + } + }, + ); +} + +class _Config extends RelayConfigNotifier { + @override + RelayConfig build() => const RelayConfig(baseUrl: 'https://tenant.example/'); +} diff --git a/mobile/test/shared/mentions/agent_policy_test.dart b/mobile/test/shared/mentions/agent_policy_test.dart index 85cdc6faef8..f195ace7e59 100644 --- a/mobile/test/shared/mentions/agent_policy_test.dart +++ b/mobile/test/shared/mentions/agent_policy_test.dart @@ -33,6 +33,9 @@ class PolicySession extends RelaySessionNotifier { final bool failPolicy; final queries = []; @override + Future fetchRelaySelf() async => + events.firstWhere((e) => e.kind == 39002).pubkey; + @override SessionState build() => const SessionState(status: SessionStatus.connected); @override Future> fetchHistory( @@ -94,9 +97,24 @@ void main() { bool failPolicy = false, void Function(PolicySession)? inspect, }) async { - final session = PolicySession(events, failPolicy: failPolicy); + final session = PolicySession([ + ...events, + signed( + owner, + 39002, + '', + tags: [ + ['d', 'channel'], + ['p', owner.public], + ['p', agent.public, '', 'bot'], + ], + ), + ], failPolicy: failPolicy); final container = ProviderContainer( - overrides: [relaySessionProvider.overrideWith(() => session)], + overrides: [ + relaySessionProvider.overrideWith(() => session), + myPubkeyProvider.overrideWithValue(owner.public), + ], ); try { final result = await container.read(agentDirectoryProvider.future);