diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs index c5fe47bd..cd224816 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs @@ -20,6 +20,8 @@ namespace GameCult.Networking private readonly CultNetDatabase _database; private readonly Func _subscribe; private readonly Func _unsubscribe; + private readonly Func? _authorizeRequest; + private readonly Func? _authorizeRecord; private readonly ICultNetSchemaServerPeerLifecycle? _peerLifecycle; private readonly ConcurrentDictionary _subscriptions = new ConcurrentDictionary(SubscriptionKeyComparer.Instance); @@ -28,10 +30,16 @@ namespace GameCult.Networking private bool _disposed; /// Attaches live database subscription handlers to a schema server. - public CultNetDatabaseSubscriptionServer(ICultNetSchemaServer server, CultNetDatabase database) + public CultNetDatabaseSubscriptionServer( + ICultNetSchemaServer server, + CultNetDatabase database, + Func? authorizeRequest = null, + Func? authorizeRecord = null) { _server = server ?? throw new ArgumentNullException(nameof(server)); _database = database ?? throw new ArgumentNullException(nameof(database)); + _authorizeRequest = authorizeRequest; + _authorizeRecord = authorizeRecord; _subscribe = HandleSubscribeAsync; _unsubscribe = HandleUnsubscribeAsync; _server.OnCultNet(_subscribe); @@ -89,6 +97,11 @@ namespace GameCult.Networking peer.SendCultNet(new CultNetErrorMessage { Error = "Database subscription requires a subscriptionId or messageId." }); return Task.CompletedTask; } + if (_authorizeRequest?.Invoke(request, peer) == false) + { + peer.SendCultNet(new CultNetErrorMessage { Error = "Database subscription is not authorized for this peer." }); + return Task.CompletedTask; + } var key = new SubscriptionKey(peer, subscriptionId); if (_requests.TryGetValue(key, out var previous)) @@ -105,7 +118,7 @@ namespace GameCult.Networking PublishDemand(request, key, active: true); if (request.IncludeSnapshot) { - peer.SendCultNet(_database.Documents.CreateRawSnapshotResponse( + var snapshot = _database.Documents.CreateRawSnapshotResponse( _database.Cache, request.MessageId, new CultNetSnapshotRequestMessage @@ -113,7 +126,15 @@ namespace GameCult.Networking MessageId = request.MessageId, SchemaIds = request.SchemaIds, RecordKeys = request.RecordKeys - })); + }); + if (_authorizeRecord != null) + { + snapshot.Documents = snapshot.Documents + .Where(document => _authorizeRecord( + request, peer, document.RecordKey, document.SchemaId)) + .ToArray(); + } + peer.SendCultNet(snapshot); } else { @@ -183,15 +204,34 @@ namespace GameCult.Networking { return _database.WatchAllChanges().Subscribe(change => { - var message = CreateChange(change, request, subscriptionId); + var message = CreateAuthorizedChange(change, request, subscriptionId, peer); if (message != null) peer.SendCultNet(message); }); } + private CultNetDatabaseChangeRawMessage? CreateAuthorizedChange( + object change, + CultNetDatabaseSubscribeMessage request, + string subscriptionId, + ICultNetSchemaServerPeer peer) + { + return CreateChangeCore(change, request, subscriptionId, (recordKey, schemaId) => + _authorizeRecord?.Invoke(request, peer, recordKey, schemaId) != false); + } + private CultNetDatabaseChangeRawMessage? CreateChange( object change, CultNetDatabaseSubscribeMessage request, string subscriptionId) + { + return CreateChangeCore(change, request, subscriptionId, (_, _) => true); + } + + private CultNetDatabaseChangeRawMessage? CreateChangeCore( + object change, + CultNetDatabaseSubscribeMessage request, + string subscriptionId, + Func authorizeRecord) { var changeType = change.GetType(); var key = (CultRecordKey)(changeType.GetProperty("Key")?.GetValue(change) ?? new CultRecordKey("")); @@ -206,6 +246,8 @@ namespace GameCult.Networking var schemaId = ResolveWireSchemaId(previous, (string?)changeType.GetProperty("SchemaId")?.GetValue(change) ?? ""); if (!MatchesRequestedSchema(request.SchemaIds, previous, schemaId)) return null; + if (!authorizeRecord(key.Value, schemaId)) + return null; return new CultNetDatabaseChangeRawMessage { MessageId = Guid.NewGuid().ToString("N"), @@ -219,6 +261,8 @@ namespace GameCult.Networking var raw = CreateRawRecord(key, document); if (!MatchesRequestedSchema(request.SchemaIds, document, raw.SchemaId)) return null; + if (!authorizeRecord(key.Value, raw.SchemaId)) + return null; return new CultNetDatabaseChangeRawMessage { MessageId = Guid.NewGuid().ToString("N"), diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 28245a3d..ae3265ca 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1462,6 +1462,81 @@ namespace GameCult.Networking.Tests Assert.That(update.Document!.RecordKey, Is.EqualTo(recordKey)); } + [Test] + public async Task DatabaseSubscriptionServer_AuthorizesRequestAndFiltersSnapshotAndLiveRecords() + { + var cache = new CultCache(); + var database = new CultNetDatabase(cache); + using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions + { + RuntimeId = "database-authorized-subscription-server", + Socket = BindUdpSocket() + }); + using var subscriptions = new CultNetDatabaseSubscriptionServer( + server, + database, + authorizeRequest: (request, _) => request.ConsumerRuntimeId == "allowed-runtime", + authorizeRecord: (_, _, recordKey, _) => recordKey.StartsWith("tests:public:", StringComparison.Ordinal)); + using var cancellation = new CancellationTokenSource(); + var serverThread = new Thread(() => + { + while (!cancellation.IsCancellationRequested) + { + _ = server.PollOnceAsync().GetAwaiter().GetResult(); + Thread.Sleep(1); + } + }) { IsBackground = true }; + serverThread.Start(); + + await database.PutAsync(new CultRecordKey("tests:public:initial"), new NetworkSchemaNote + { + Schema = "tests.networking_note.v1", + Text = "public-initial" + }); + await database.PutAsync(new CultRecordKey("tests:private:initial"), new NetworkSchemaNote + { + Schema = "tests.networking_note.v1", + Text = "private-initial" + }); + + using var client = CultNetSchemaClients.CreateRudp("database-authorized-subscription-client"); + var subscribed = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + var changed = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + client.OnCultNet(message => subscribed.TrySetResult(message)); + client.OnCultNet(message => changed.TrySetResult(message)); + client.Connect("127.0.0.1", server.LocalEndPoint.Port); + await WaitUntilAsync(() => client.Connected, TimeSpan.FromSeconds(2)); + client.SendCultNet(new CultNetDatabaseSubscribeMessage + { + MessageId = "subscribe-authorized", + SubscriptionId = "authorized", + ConsumerRuntimeId = "allowed-runtime", + IncludeSnapshot = true + }); + + var snapshot = await AwaitWithTimeout(subscribed.Task, TimeSpan.FromSeconds(2)); + Assert.That(snapshot.Documents.Select(document => document.RecordKey), + Is.EqualTo(new[] { "tests:public:initial" })); + + await database.PutAsync(new CultRecordKey("tests:private:live"), new NetworkSchemaNote + { + Schema = "tests.networking_note.v1", + Text = "private-live" + }); + await database.PutAsync(new CultRecordKey("tests:public:live"), new NetworkSchemaNote + { + Schema = "tests.networking_note.v1", + Text = "public-live" + }); + var update = await AwaitWithTimeout(changed.Task, TimeSpan.FromSeconds(2)); + + cancellation.Cancel(); + Assert.That(update.Document, Is.Not.Null); + Assert.That(update.Document!.RecordKey, Is.EqualTo("tests:public:live")); + } + [Test] public async Task DatabaseSubscriptionServer_ProjectsBodyDemandAndWithdrawalFromExactSubscription() {