diff --git a/contracts/cultnet/cultnet-distributed-database.md b/contracts/cultnet/cultnet-distributed-database.md index 66bb450e..a170835e 100644 --- a/contracts/cultnet/cultnet-distributed-database.md +++ b/contracts/cultnet/cultnet-distributed-database.md @@ -111,8 +111,9 @@ Not implemented yet: not guess. Adding a code is therefore a coordinated change across every runtime, made together: the schema (both copies), the C# `CultNetErrorMessage` factory, the Rust `CultNetErrorCode`, and a captured-bytes case in `packages/cultnet-rs/tests/error_message.rs` for it. TypeScript reads the schema. The current codes are -`selection_invalid`, `cursor_stale`, `cursor_invalid`, `reference_outside_target`, `variant_unsupported` and -`unowned_schema` (a write, or a remote put or delete, for a schema and key no shard owns). +`selection_invalid`, `cursor_stale`, `cursor_invalid`, `reference_outside_target`, `variant_unsupported`, +`unowned_schema` (a write, or a remote put or delete, for a schema and key no shard owns) and +`write_not_authorized` (a peer's put or delete its server's remote write authority did not grant). ## Live Invariants @@ -122,6 +123,9 @@ copies), the C# `CultNetErrorMessage` factory, the Rust `CultNetErrorCode`, and - One primary owns writes for a shard until a stronger policy exists. - Non-owner writes are rejected or explicitly forwarded. They are not silently applied. +- A peer's put or delete is applied or forwarded only when the server's + `RemoteWriteAuthority` grants it; none configured refuses every remote write. + The connection key admits a peer to the transport and is never a grant. - Raw snapshot/put/delete messages must pass through `CultNetDatabase`, not bypass shard policy. - The accepted shard log is stored in replica catch-up wire form. Object logs diff --git a/contracts/cultnet/cultnet.error.schema.json b/contracts/cultnet/cultnet.error.schema.json index 59764338..e3b94c2b 100644 --- a/contracts/cultnet/cultnet.error.schema.json +++ b/contracts/cultnet/cultnet.error.schema.json @@ -28,7 +28,7 @@ "code": { "description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it. The set is a closed enum: Rust and TypeScript refuse a message whose code is not listed, so adding a code is one coordinated change across every runtime (this schema, the C# factory, the Rust CultNetErrorCode, and a captured-bytes case for each).", "type": ["string", "null"], - "enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema"] + "enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema", "write_not_authorized"] }, "details": { "description": "Code-specific structured detail (R-N). Every field is populated only for the code that defines it; the others are null, not omitted.", @@ -36,8 +36,8 @@ "additionalProperties": false, "required": ["field", "value", "asOf", "current"], "properties": { - "field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended. unowned_schema: \"schemaId\".", "type": ["string", "null"] }, - "value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended. unowned_schema: the schema id no shard owns.", "type": ["string", "null"] }, + "field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended. unowned_schema: \"schemaId\". write_not_authorized: \"recordKey\".", "type": ["string", "null"] }, + "value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended. unowned_schema: the schema id no shard owns. write_not_authorized: the key of the refused write.", "type": ["string", "null"] }, "asOf": { "description": "cursor_stale: the cursor's own minted asOf.", "type": ["integer", "null"], "minimum": 0 }, "current": { "description": "cursor_stale: the answering server's current asOf.", "type": ["integer", "null"], "minimum": 0 } } diff --git a/packages/cultnet-rs/contracts/cultnet/cultnet.error.schema.json b/packages/cultnet-rs/contracts/cultnet/cultnet.error.schema.json index 59764338..e3b94c2b 100644 --- a/packages/cultnet-rs/contracts/cultnet/cultnet.error.schema.json +++ b/packages/cultnet-rs/contracts/cultnet/cultnet.error.schema.json @@ -28,7 +28,7 @@ "code": { "description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it. The set is a closed enum: Rust and TypeScript refuse a message whose code is not listed, so adding a code is one coordinated change across every runtime (this schema, the C# factory, the Rust CultNetErrorCode, and a captured-bytes case for each).", "type": ["string", "null"], - "enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema"] + "enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema", "write_not_authorized"] }, "details": { "description": "Code-specific structured detail (R-N). Every field is populated only for the code that defines it; the others are null, not omitted.", @@ -36,8 +36,8 @@ "additionalProperties": false, "required": ["field", "value", "asOf", "current"], "properties": { - "field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended. unowned_schema: \"schemaId\".", "type": ["string", "null"] }, - "value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended. unowned_schema: the schema id no shard owns.", "type": ["string", "null"] }, + "field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended. unowned_schema: \"schemaId\". write_not_authorized: \"recordKey\".", "type": ["string", "null"] }, + "value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended. unowned_schema: the schema id no shard owns. write_not_authorized: the key of the refused write.", "type": ["string", "null"] }, "asOf": { "description": "cursor_stale: the cursor's own minted asOf.", "type": ["integer", "null"], "minimum": 0 }, "current": { "description": "cursor_stale: the answering server's current asOf.", "type": ["integer", "null"], "minimum": 0 } } diff --git a/packages/cultnet-rs/src/contracts.rs b/packages/cultnet-rs/src/contracts.rs index e4f67164..56afaa7c 100644 --- a/packages/cultnet-rs/src/contracts.rs +++ b/packages/cultnet-rs/src/contracts.rs @@ -246,6 +246,7 @@ pub enum CultNetErrorCode { ReferenceOutsideTarget, VariantUnsupported, UnownedSchema, + WriteNotAuthorized, } impl CultNetErrorCode { @@ -257,6 +258,7 @@ impl CultNetErrorCode { Self::ReferenceOutsideTarget => "reference_outside_target", Self::VariantUnsupported => "variant_unsupported", Self::UnownedSchema => "unowned_schema", + Self::WriteNotAuthorized => "write_not_authorized", } } @@ -268,6 +270,7 @@ impl CultNetErrorCode { "reference_outside_target" => Ok(Self::ReferenceOutsideTarget), "variant_unsupported" => Ok(Self::VariantUnsupported), "unowned_schema" => Ok(Self::UnownedSchema), + "write_not_authorized" => Ok(Self::WriteNotAuthorized), _ => Err(anyhow!("ErrorMessage.Code is not a recognized cultnet.error.v0 code")), } } diff --git a/packages/cultnet-rs/tests/error_message.rs b/packages/cultnet-rs/tests/error_message.rs index 5c633087..58732295 100644 --- a/packages/cultnet-rs/tests/error_message.rs +++ b/packages/cultnet-rs/tests/error_message.rs @@ -158,6 +158,21 @@ fn cases() -> Vec { }), }, }, + Case { + name: "write_not_authorized", + // C#: ForRemoteWriteRefused(new CultNetRemoteWriteRequest(null, Put, "some.schema.v1", new CultRecordKey("some:key"), null, null), "not granted") + csharp_hex: "85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72D92177726974655F6E6F745F617574686F72697A65643A206E6F74206772616E746564AB726F7574696E6748696E74C0A4636F6465B477726974655F6E6F745F617574686F72697A6564A764657461696C7384A56669656C64A97265636F72644B6579A576616C7565A8736F6D653A6B6579A461734F66C0A763757272656E74C0", + message: || CultNetMessage::Error { + error: "write_not_authorized: not granted".to_string(), + code: Some(CultNetErrorCode::WriteNotAuthorized), + details: Some(CultNetErrorDetails { + field: Some("recordKey".to_string()), + value: Some("some:key".to_string()), + as_of: None, + current: None, + }), + }, + }, ] } diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index 668c088e..d9e7ec09 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -1015,11 +1015,16 @@ namespace GameCult.Networking } /// - /// Applies a raw document mutation after checking shard authority. + /// Applies a raw document mutation after checking shard authority. This is the host's own door (and replication's): + /// it asks no remote write authority. A peer's put goes through , which does. /// /// No shard owns the schema and key; nothing was written. /// The change committed and was published, but the shard log could not record it. - public async Task ApplyPutAsync(CultNetDocumentPutRawMessage message) + public Task ApplyPutAsync(CultNetDocumentPutRawMessage message) => ApplyPutAsync(message, authorize: null); + + // The peer door: the same apply, with the server's authorize callback asked once the schema is resolved and before the + // shard is. The callback throws to refuse. The public overload is the host's own door and asks no remote authority. + internal async Task ApplyPutAsync(CultNetDocumentPutRawMessage message, Action? authorize) { ThrowIfDisposed(); if (message == null) throw new ArgumentNullException(nameof(message)); @@ -1030,6 +1035,7 @@ namespace GameCult.Networking var key = new CultRecordKey(message.Document.RecordKey); var descriptor = _documents.ResolveDescriptorForRawDocument(message.Document); + authorize?.Invoke(descriptor.SchemaId, key); var shard = ResolveShardInternal(descriptor, key); EnsurePrimary(shard, descriptor.SchemaId, key, message.ShardEpoch); var document = _documents.DeserializeRawDocument(message.Document); @@ -1039,17 +1045,22 @@ namespace GameCult.Networking } /// - /// Applies a raw document delete after checking shard authority. + /// Applies a raw document delete after checking shard authority. This is the host's own door: it asks no remote + /// write authority. A peer's delete goes through , which does. /// /// No shard owns the schema and key; nothing was written. /// The change committed and was published, but the shard log could not record it. - public Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message) + public Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message) => ApplyDeleteAsync(message, authorize: null); + + // The peer door; see the put overload. + internal Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message, Action? authorize) { ThrowIfDisposed(); if (message == null) throw new ArgumentNullException(nameof(message)); var key = new CultRecordKey(message.RecordKey); var descriptor = _documents.ResolveDescriptorForSchemaId(message.SchemaId); + authorize?.Invoke(descriptor.SchemaId, key); var shard = ResolveShardInternal(descriptor, key); EnsurePrimary(shard, descriptor.SchemaId, key, message.ShardEpoch); var previous = _cache.Get(key); diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index 77b03070..149b9619 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -67,22 +67,6 @@ namespace GameCult.Networking /// public CultNetDatabase Database => _database; - /// - /// Applies a raw put message through the database shard policy. - /// - public Task ApplyPutAsync(CultNetDocumentPutRawMessage message) - { - return _database.ApplyPutAsync(message); - } - - /// - /// Applies a raw delete message through the database shard policy. - /// - public Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message) - { - return _database.ApplyDeleteAsync(message); - } - /// /// Creates a raw snapshot response from the database cache. /// @@ -304,11 +288,52 @@ namespace GameCult.Networking return Task.CompletedTask; } + // The one question asked of the write authority per received write. None configured refuses; an authority that throws + // refuses too (fail closed) and the fault is logged as an error. + private Action Authorize( + CultNetServerPeer peer, + CultNetRemoteWriteOperation operation, + string? shardId, + long? shardEpoch) + { + return (schemaId, key) => + { + var request = new CultNetRemoteWriteRequest(peer, operation, schemaId, key, shardId, shardEpoch); + var authority = _options.RemoteWriteAuthority; + if (authority == null) + { + throw new CultNetRemoteWriteRefusedException(request, "no remote write authority is configured"); + } + + CultNetRemoteWriteDecision decision; + try + { + decision = authority.Decide(request); + } + catch (Exception ex) + { + _server.Logger.LogError($"CultNet remote write authority failed: {ex.Message}"); + throw new CultNetRemoteWriteRefusedException(request, "the write authority failed"); + } + + if (decision == null || !decision.IsGranted) + { + throw new CultNetRemoteWriteRefusedException(request, decision?.Reason); + } + }; + } + private async Task HandlePutAsync(CultNetDocumentPutRawMessage message, CultNetServerPeer peer) { try { - await ApplyPutAsync(message).ConfigureAwait(false); + await _database.ApplyPutAsync(message, Authorize(peer, CultNetRemoteWriteOperation.Put, message.ShardId, message.ShardEpoch)) + .ConfigureAwait(false); + } + catch (CultNetRemoteWriteRefusedException ex) + { + _server.Logger.LogWarning($"CultNet raw put refused: {ex.Message}"); + peer.SendCultNet(CultNetErrorMessage.ForRemoteWriteRefused(ex.Request, ex.Reason)); } catch (CultNetShardAuthorityException ex) { @@ -339,7 +364,13 @@ namespace GameCult.Networking { try { - await ApplyDeleteAsync(message).ConfigureAwait(false); + await _database.ApplyDeleteAsync(message, Authorize(peer, CultNetRemoteWriteOperation.Delete, message.ShardId, message.ShardEpoch)) + .ConfigureAwait(false); + } + catch (CultNetRemoteWriteRefusedException ex) + { + _server.Logger.LogWarning($"CultNet raw delete refused: {ex.Message}"); + peer.SendCultNet(CultNetErrorMessage.ForRemoteWriteRefused(ex.Request, ex.Reason)); } catch (CultNetShardAuthorityException ex) { diff --git a/src/GameCult.Networking/CultNetRemoteWriteAuthority.cs b/src/GameCult.Networking/CultNetRemoteWriteAuthority.cs new file mode 100644 index 00000000..b3083f19 --- /dev/null +++ b/src/GameCult.Networking/CultNetRemoteWriteAuthority.cs @@ -0,0 +1,113 @@ +using System; +using GameCult.Caching; + +namespace GameCult.Networking +{ + /// + /// Decides whether a connected peer may put or delete a record in the database a + /// serves. Transport admission (the connection key, a completed handshake) is + /// never a grant; this is the one owner of that decision. + /// + public interface ICultNetRemoteWriteAuthority + { + /// + /// Decides one peer write. Called once per received put or delete, after the schema is resolved and before the + /// shard is, so a refusal applies, publishes, logs and forwards nothing. + /// + CultNetRemoteWriteDecision Decide(CultNetRemoteWriteRequest request); + } + + /// + /// The kind of write a peer asked for. + /// + public enum CultNetRemoteWriteOperation + { + /// Put a record. + Put, + /// Delete a record. + Delete + } + + /// + /// One peer write, as the authority sees it. + /// + public sealed class CultNetRemoteWriteRequest + { + /// + /// Creates a request. + /// + public CultNetRemoteWriteRequest( + CultNetServerPeer peer, + CultNetRemoteWriteOperation operation, + string schemaId, + CultRecordKey recordKey, + string? shardId, + long? shardEpoch) + { + Peer = peer; + Operation = operation; + SchemaId = schemaId; + RecordKey = recordKey; + ShardId = shardId; + ShardEpoch = shardEpoch; + } + + /// Gets the peer connection the message arrived on (endpoint and transport only). + public CultNetServerPeer Peer { get; } + + /// Gets the operation. + public CultNetRemoteWriteOperation Operation { get; } + + /// Gets the canonical schema id the database resolved for the write, never the alias the peer sent. + public string SchemaId { get; } + + /// Gets the record key. + public CultRecordKey RecordKey { get; } + + /// Gets the shard id as the peer sent it. + public string? ShardId { get; } + + /// Gets the shard epoch as the peer sent it. + public long? ShardEpoch { get; } + } + + /// + /// An authority's answer to one . + /// + public sealed class CultNetRemoteWriteDecision + { + private CultNetRemoteWriteDecision(bool isGranted, string? reason) + { + IsGranted = isGranted; + Reason = reason; + } + + /// Gets whether the write may proceed. + public bool IsGranted { get; } + + /// Gets the refusal reason, when there is one. + public string? Reason { get; } + + /// Grants the write. + public static CultNetRemoteWriteDecision Grant() => new(true, null); + + /// Refuses the write. + public static CultNetRemoteWriteDecision Refuse(string? reason = null) => new(false, reason); + } + + // Thrown by the server's authorize callback through the database's remote apply when the authority refuses; the + // server's handlers are its one catcher. + internal sealed class CultNetRemoteWriteRefusedException : Exception + { + public CultNetRemoteWriteRefusedException(CultNetRemoteWriteRequest request, string? reason) + : base(reason ?? "the write was not authorized") + { + Request = request; + Reason = reason; + } + + public CultNetRemoteWriteRequest Request { get; } + + public string? Reason { get; } + } +} diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index e7f4cfd9..baf0b6d5 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -511,7 +511,7 @@ namespace GameCult.Networking [Key("routingHint")] public CultNetShardRoutingHint? RoutingHint { get; set; } /// /// Gets or sets the machine-readable refusal code (R-N), e.g. selection_invalid, - /// cursor_stale, cursor_invalid, reference_outside_target, unowned_schema. Additive to + /// cursor_stale, cursor_invalid, reference_outside_target, unowned_schema, write_not_authorized. Additive to /// cultnet.error.v0; a v0 peer that does not know this field ignores it. /// [Key("code")] public string? Code { get; set; } @@ -543,6 +543,18 @@ namespace GameCult.Networking Details = new CultNetErrorDetails { Field = "schemaId", Value = ex.SchemaId } }; + /// + /// Builds the wire error for a peer write its authority refused: code write_not_authorized, details + /// { field: "recordKey", value }. + /// + public static CultNetErrorMessage ForRemoteWriteRefused(CultNetRemoteWriteRequest request, string? reason) => + new CultNetErrorMessage + { + Error = string.IsNullOrWhiteSpace(reason) ? "write_not_authorized" : $"write_not_authorized: {reason}", + Code = "write_not_authorized", + Details = new CultNetErrorDetails { Field = "recordKey", Value = request.RecordKey.Value } + }; + /// /// Builds the wire error for a cursor refusal (R-N): code cursor_stale with details /// { asOf, current }, or code cursor_invalid with no details. diff --git a/src/GameCult.Networking/CultNetShardWriteForwarding.cs b/src/GameCult.Networking/CultNetShardWriteForwarding.cs index b2b9e7a2..7038b0a7 100644 --- a/src/GameCult.Networking/CultNetShardWriteForwarding.cs +++ b/src/GameCult.Networking/CultNetShardWriteForwarding.cs @@ -34,6 +34,13 @@ namespace GameCult.Networking /// Gets or sets the forwarder used when non-primary write forwarding is enabled. /// public ICultNetShardWriteForwarder? WriteForwarder { get; set; } + + /// + /// Gets or sets the authority asked for every peer put and delete. None configured refuses them all: a connection + /// key admits a peer to the transport, never to write. A write forwarded to a primary is judged here for the + /// original peer and again at the primary for the forwarder's connection. + /// + public ICultNetRemoteWriteAuthority? RemoteWriteAuthority { get; set; } } /// diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index de8496d3..7347d145 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -197,7 +197,10 @@ reply. The first coherent shard policy is primary ownership. Each shard has one authoritative writer at a time. Clients may connect to any node; a non-owner node forwards a write to the owner or rejects it honestly when forwarding is -not available. Followers subscribe to committed shard mutations and apply them +not available. A node applies or forwards a peer's write only after its +`RemoteWriteAuthority` grants it; with none configured it refuses all of them, and a +write forwarded to a primary is judged again there for the forwarder's connection. +Followers subscribe to committed shard mutations and apply them through the same CultCache reconciliation path as local file changes. This should feel like the local-first parts of Firebase or RethinkDB without diff --git a/tests/GameCult.Networking.Tests/CultNetJournalTests.cs b/tests/GameCult.Networking.Tests/CultNetJournalTests.cs index 6285d674..eb68b1fc 100644 --- a/tests/GameCult.Networking.Tests/CultNetJournalTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetJournalTests.cs @@ -733,7 +733,7 @@ namespace GameCult.Networking.Tests using var server = new Server(cache, ServerSecurityOptions.Development()); var logger = new CapturingLogger(); server.Logger = logger; - using var databaseServer = new CultNetDatabaseServer(server, database); + using var databaseServer = new CultNetDatabaseServer(server, database, new CultNetDatabaseServerOptions { RemoteWriteAuthority = new TestGrantsEverything() }); var handler = (Func)typeof(CultNetDatabaseServer) .GetField("_putHandler", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)! .GetValue(databaseServer)!; @@ -929,7 +929,7 @@ namespace GameCult.Networking.Tests var host = new Server(cache, ServerSecurityOptions.Development()); var logger = new CapturingLogger(); host.Logger = logger; - var databaseServer = new CultNetDatabaseServer(host, database); + var databaseServer = new CultNetDatabaseServer(host, database, new CultNetDatabaseServerOptions { RemoteWriteAuthority = new TestGrantsEverything() }); var peer = (CultNetServerPeer)System.Runtime.CompilerServices.RuntimeHelpers.GetUninitializedObject(typeof(CultNetServerPeer)); const System.Reflection.BindingFlags flags = System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance; var put = (Func)typeof(CultNetDatabaseServer).GetField("_putHandler", flags)!.GetValue(databaseServer)!; diff --git a/tests/GameCult.Networking.Tests/CultNetSchemaContractTests.cs b/tests/GameCult.Networking.Tests/CultNetSchemaContractTests.cs index 3aa2705d..8b989bb6 100644 --- a/tests/GameCult.Networking.Tests/CultNetSchemaContractTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetSchemaContractTests.cs @@ -206,6 +206,20 @@ namespace GameCult.Networking.Tests AssertErrorMatchesSchema(message); } + [Test] + public void ErrorMessage_RemoteWriteRefusedRealBytes_MatchesItsSchema() + { + var request = new CultNetRemoteWriteRequest(null!, CultNetRemoteWriteOperation.Put, "some.schema.v1", new CultRecordKey("some:key"), null, null); + var message = CultNetErrorMessage.ForRemoteWriteRefused(request, "not granted"); + + Assert.That(message.Code, Is.EqualTo("write_not_authorized")); + Assert.That(message.Details!.Field, Is.EqualTo("recordKey")); + Assert.That(message.Details.Value, Is.EqualTo("some:key")); + AssertErrorMatchesSchema(message); + // The bytes cultnet-rs's error_message.rs write_not_authorized case decodes and re-encodes. + Assert.That(Convert.ToHexString(CultNetSchemaMessageSerialization.Serialize(message)), Is.EqualTo("85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72D92177726974655F6E6F745F617574686F72697A65643A206E6F74206772616E746564AB726F7574696E6748696E74C0A4636F6465B477726974655F6E6F745F617574686F72697A6564A764657461696C7384A56669656C64A97265636F72644B6579A576616C7565A8736F6D653A6B6579A461734F66C0A763757272656E74C0")); + } + private static void AssertErrorMatchesSchema(CultNetErrorMessage message) { var validator = new MiniJsonSchemaValidator(SchemaDir()); diff --git a/tests/GameCult.Networking.Tests/RemoteWriteAuthorityTests.cs b/tests/GameCult.Networking.Tests/RemoteWriteAuthorityTests.cs new file mode 100644 index 00000000..897674b1 --- /dev/null +++ b/tests/GameCult.Networking.Tests/RemoteWriteAuthorityTests.cs @@ -0,0 +1,439 @@ +#nullable enable +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Net; +using System.Reflection; +using System.Runtime.CompilerServices; +using System.Threading.Tasks; +using GameCult.Caching; +using LiteNetLib; +using NUnit.Framework; +using R3; +using static GameCult.Networking.Tests.NetworkingTests; + +namespace GameCult.Networking.Tests +{ + // A peer's put or delete is applied or forwarded only when the server's RemoteWriteAuthority grants it. The wire handlers + // are reached through their private delegates with an uninitialized peer, as CultNetJournalTests.WireHandlers does: + // anything the handler tries to send throws on that peer, so a handler that throws answered the peer with a refusal and a + // handler that returns quietly sent nothing. + // The tests' own grant-everyone authority, for tests whose subject is not the grant. The library ships none. + internal sealed class TestGrantsEverything : ICultNetRemoteWriteAuthority + { + public CultNetRemoteWriteDecision Decide(CultNetRemoteWriteRequest request) => CultNetRemoteWriteDecision.Grant(); + } + + public sealed class RemoteWriteAuthorityTests + { + private const string ShardId = "authority-shard"; + private static readonly CultRecordKey KeyA = new("tests:authority:a"); + private static readonly CultRecordKey KeyB = new("tests:authority:b"); + private static readonly CultRecordKey Unowned = new("unrelated:authority"); + + private static NetworkSchemaNote Note(string text) => new() { Schema = "tests.networking_note.v1", Text = text }; + + private sealed class Rig + { + public CultCache Cache = null!; + public CultNetDatabase Database = null!; + public CultNetDatabaseServer DatabaseServer = null!; + public Server Host = null!; + public CultNetServerPeer Peer = null!; + public CapturingLogger Logger = null!; + public Func Put = null!; + public Func Delete = null!; + public string SchemaId = string.Empty; + public List Published = new(); + public IDisposable Subscription = null!; + + public string? TextOf(CultRecordKey key) => Cache.Get(key)?.Text; + public int LogLength => Database.GetMutationLog(ShardId).Count; + + public CultNetDocumentPutRawMessage PutOf(CultRecordKey key, string text) + { + var message = Database.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle(key), Note(text)); + message.ShardId = ShardId; + message.ShardEpoch = 1; + return message; + } + + public CultNetDocumentDeleteMessage DeleteOf(CultRecordKey key) => + new() { MessageId = "wire-delete", SchemaId = SchemaId, RecordKey = key.Value, ShardId = ShardId, ShardEpoch = 1 }; + } + + private static Rig Build(ICultNetRemoteWriteAuthority? authority, bool primary = true, ICultNetShardWriteForwarder? forwarder = null, CultNetServerPeer? peer = null) + { + var cache = new CultCache(); + var schemaId = cache.Registry.GetRequired().SchemaId; + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + Shards = + [ + new CultNetShardDescriptor(ShardId, "primary", epoch: 1, isPrimary: primary, schemaIds: [schemaId], + primaryEndpoints: primary ? null : ["cultnet://runtime-owner:3075"]) + ] + }); + var host = new Server(cache, ServerSecurityOptions.Development()); + var logger = new CapturingLogger(); + host.Logger = logger; + var databaseServer = new CultNetDatabaseServer(host, database, new CultNetDatabaseServerOptions + { + RemoteWriteAuthority = authority, + ForwardNonPrimaryWrites = forwarder != null, + WriteForwarder = forwarder + }); + peer ??= (CultNetServerPeer)RuntimeHelpers.GetUninitializedObject(typeof(CultNetServerPeer)); + const BindingFlags flags = BindingFlags.NonPublic | BindingFlags.Instance; + var put = (Func)typeof(CultNetDatabaseServer).GetField("_putHandler", flags)!.GetValue(databaseServer)!; + var delete = (Func)typeof(CultNetDatabaseServer).GetField("_deleteHandler", flags)!.GetValue(databaseServer)!; + var rig = new Rig + { + Cache = cache, + Database = database, + DatabaseServer = databaseServer, + Host = host, + Peer = peer, + Logger = logger, + SchemaId = schemaId, + Put = message => put(message, peer!), + Delete = message => delete(message, peer!) + }; + rig.Subscription = database.WatchAllChanges().Subscribe(change => rig.Published.Add(change)); + return rig; + } + + // The handler answered the peer with a refusal: the send reached the (uninitialized) peer and failed there, so the + // refusal message was built first. A failure earlier, in building it, would not pass through the peer's send. + private static void AssertAnswered(Func handler, string because) + { + var failure = Assert.CatchAsync(async () => await handler(), because)!; + Assert.That(failure.ToString(), Does.Contain("CultNetServerPeer.SendCultNet").And.Not.Contain("ForRemoteWriteRefused"), because); + } + + // The server writes its own record through the host's door, which asks no remote authority. + private static async Task Seed(Rig rig, CultRecordKey key, string text) => await rig.Database.ApplyPutAsync(rig.PutOf(key, text)); + + [Test] + public async Task WithNoAuthorityAPeersPutAndDeleteAreRefusedAndChangeNothing() + { + var rig = Build(authority: null); + await Seed(rig, KeyA, "server"); + var publishedBefore = rig.Published.Count; + var logBefore = rig.LogLength; + Assert.That(rig.TextOf(KeyA), Is.EqualTo("server"), "the host's own door needs no remote authority"); + + AssertAnswered(() => rig.Put(rig.PutOf(KeyA, "peer")), "the peer is answered with a refusal"); + AssertAnswered(() => rig.Delete(rig.DeleteOf(KeyA)), "the peer is answered with a refusal"); + AssertAnswered(() => rig.Put(rig.PutOf(KeyB, "peer")), "a new key is refused too"); + + Assert.That(rig.TextOf(KeyA), Is.EqualTo("server")); + Assert.That(rig.TextOf(KeyB), Is.Null); + Assert.That(rig.Published, Has.Count.EqualTo(publishedBefore), "a refusal publishes no change"); + Assert.That(rig.LogLength, Is.EqualTo(logBefore), "a refusal appends no mutation log entry"); + Assert.That(rig.Logger.Warnings, Has.Some.Contains("raw put refused")); + Assert.That(rig.Logger.Warnings, Has.Some.Contains("raw delete refused")); + Assert.That(rig.Logger.Errors, Is.Empty, "a refusal is not a fault"); + } + + [Test] + public async Task AGrantedPutAndDeleteCommitAndTheAuthoritySawThisPeer() + { + var authority = new RecordingAuthority(_ => CultNetRemoteWriteDecision.Grant()); + var rig = Build(authority); + + await rig.Put(rig.PutOf(KeyA, "peer")); + Assert.That(rig.TextOf(KeyA), Is.EqualTo("peer")); + Assert.That(rig.Published, Has.Count.EqualTo(1)); + Assert.That(rig.LogLength, Is.EqualTo(1)); + + await rig.Delete(rig.DeleteOf(KeyA)); + Assert.That(rig.TextOf(KeyA), Is.Null); + + Assert.That(authority.Requests, Has.Count.EqualTo(2), "asked once per received write"); + Assert.That(authority.Requests.Select(request => request.Peer), Has.All.SameAs(rig.Peer)); + Assert.That(rig.Logger.Errors, Is.Empty); + } + + [Test] + public async Task TheRequestCarriesTheResolvedWriteAndScopedGrantsRefuseTheirNeighbours() + { + // The put names a foreign alias of the schema; the payload routes it to the canonical descriptor. + var recorded = new RecordingAuthority(request => + request.Operation == CultNetRemoteWriteOperation.Put && request.RecordKey.Equals(KeyA) + ? CultNetRemoteWriteDecision.Grant() + : CultNetRemoteWriteDecision.Refuse("scoped")); + var rig = Build(recorded); + await Seed(rig, KeyA, "server"); + var aliased = rig.PutOf(KeyA, "aliased"); + aliased.Document.SchemaId = "runtime.generated.network-note.ui.42"; + + await rig.Put(aliased); + + var request = recorded.Requests.Single(); + Assert.That(request.SchemaId, Is.EqualTo(rig.SchemaId), "the authority sees the canonical id, not the alias as sent"); + Assert.That(request.Operation, Is.EqualTo(CultNetRemoteWriteOperation.Put)); + Assert.That(request.RecordKey, Is.EqualTo(KeyA)); + Assert.That(request.ShardId, Is.EqualTo(ShardId)); + Assert.That(request.ShardEpoch, Is.EqualTo(1)); + Assert.That(rig.TextOf(KeyA), Is.EqualTo("aliased")); + + AssertAnswered(() => rig.Put(rig.PutOf(KeyB, "other-key")), "the peer is answered with a refusal"); + AssertAnswered(() => rig.Delete(rig.DeleteOf(KeyA)), "the peer is answered with a refusal"); + Assert.That(rig.TextOf(KeyB), Is.Null, "a grant scoped to key a refuses key b"); + Assert.That(rig.TextOf(KeyA), Is.EqualTo("aliased"), "a grant scoped to Put refuses Delete"); + Assert.That(recorded.Requests.Last().Operation, Is.EqualTo(CultNetRemoteWriteOperation.Delete)); + Assert.That(recorded.Requests.Last().SchemaId, Is.EqualTo(rig.SchemaId)); + } + + [Test] + public async Task ARefusedWriteIsNeverForwardedAndAGrantedOneIs() + { + var forwarder = new CountingForwarder(); + var refusing = Build(new RecordingAuthority(_ => CultNetRemoteWriteDecision.Refuse("no")), primary: false, forwarder); + + AssertAnswered(() => refusing.Put(refusing.PutOf(KeyA, "peer")), "the peer is answered with a refusal"); + AssertAnswered(() => refusing.Delete(refusing.DeleteOf(KeyA)), "the peer is answered with a refusal"); + Assert.That(forwarder.Puts, Is.EqualTo(0), "a refused put is not forwarded"); + Assert.That(forwarder.Deletes, Is.EqualTo(0), "a refused delete is not forwarded"); + + var unconfigured = Build(authority: null, primary: false, forwarder); + AssertAnswered(() => unconfigured.Put(unconfigured.PutOf(KeyA, "peer")), "the peer is answered with a refusal"); + Assert.That(forwarder.Puts, Is.EqualTo(0), "no authority forwards nothing"); + + var granting = Build(new RecordingAuthority(_ => CultNetRemoteWriteDecision.Grant()), primary: false, forwarder); + await granting.Put(granting.PutOf(KeyA, "peer")); + await granting.Delete(granting.DeleteOf(KeyA)); + Assert.That(forwarder.Puts, Is.EqualTo(1), "a granted write on a non-primary shard is forwarded as before"); + Assert.That(forwarder.Deletes, Is.EqualTo(1)); + } + + [Test] + public void RefusalPrecedesTheUnownedSchemaCheck() + { + var unowned = new MeshQuickstartNote { NoteId = "u", Body = "unowned" }; + var rig = Build(authority: null); + var put = rig.Database.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle(Unowned), unowned); + put.ShardId = ShardId; + put.ShardEpoch = 1; + + AssertAnswered(() => rig.Put(put), "the peer is answered with a refusal"); + + Assert.That(rig.Logger.Warnings, Has.Some.Contains("authority"), "an ungranted put of an unowned schema is answered write_not_authorized"); + Assert.That(rig.Logger.Warnings, Has.None.Contains("No shard owns")); + + var granting = Build(new RecordingAuthority(_ => CultNetRemoteWriteDecision.Grant())); + var grantedPut = granting.Database.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle(Unowned), unowned); + grantedPut.ShardId = ShardId; + grantedPut.ShardEpoch = 1; + AssertAnswered(() => granting.Put(grantedPut), "the peer is answered with a refusal"); + Assert.That(granting.Logger.Warnings, Has.Some.Contains("No shard owns"), "a granted put still meets the unowned-schema refusal"); + } + + [Test] + public async Task AnAuthorityThatThrowsRefusesAndIsLoggedAsAnError() + { + var rig = Build(new RecordingAuthority(_ => throw new InvalidOperationException("authority exploded"))); + await Seed(rig, KeyA, "server"); + + AssertAnswered(() => rig.Put(rig.PutOf(KeyA, "peer")), "the peer is answered with a refusal"); + AssertAnswered(() => rig.Delete(rig.DeleteOf(KeyA)), "the peer is answered with a refusal"); + + Assert.That(rig.TextOf(KeyA), Is.EqualTo("server"), "fail closed"); + Assert.That(rig.Logger.Errors, Has.Some.Contains("authority exploded")); + } + + [Test] + public void ARefusalWithNoReasonStillSaysWhy() + { + var rig = Build(new RecordingAuthority(_ => CultNetRemoteWriteDecision.Refuse())); + + AssertAnswered(() => rig.Put(rig.PutOf(KeyA, "peer")), "refused"); + + Assert.That(rig.Logger.Warnings, Has.Some.Contains("raw put refused: the write was not authorized")); + } + + [Test] + public void TheRefusalNamesTheKeyAndTheCode() + { + var request = new CultNetRemoteWriteRequest(null!, CultNetRemoteWriteOperation.Delete, "s", KeyA, null, null); + var message = CultNetErrorMessage.ForRemoteWriteRefused(request, "why not"); + + Assert.That(message.Code, Is.EqualTo("write_not_authorized")); + Assert.That(message.Error, Is.EqualTo("write_not_authorized: why not")); + Assert.That(message.Details!.Field, Is.EqualTo("recordKey")); + Assert.That(message.Details.Value, Is.EqualTo(KeyA.Value)); + Assert.That(message.RoutingHint, Is.Null); + Assert.That(CultNetErrorMessage.ForRemoteWriteRefused(request, null).Error, Is.EqualTo("write_not_authorized")); + } + + // A real LiteNetLib pair on loopback with both sockets bound to ephemeral ports (Server and Client bind fixed ports, + // which a parallel suite cannot share). The server side is a real NetPeer wrapped as the CultNetServerPeer the handlers + // answer; the client side decodes what arrives on the wire with the production schema decoder. + private sealed class WirePair : IDisposable + { + private readonly NetManager _serverManager; + private readonly NetManager _clientManager; + public CultNetServerPeer Peer { get; } + public ConcurrentQueue Errors { get; } = new(); + + public WirePair() + { + var serverListener = new EventBasedNetListener(); + _serverManager = new NetManager(serverListener) { UnsyncedEvents = true }; + var clientListener = new EventBasedNetListener(); + _clientManager = new NetManager(clientListener) { UnsyncedEvents = true }; + var connected = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + serverListener.ConnectionRequestEvent += request => request.AcceptIfKey("rwa-wire"); + serverListener.PeerConnectedEvent += serverSide => connected.TrySetResult(serverSide); + var decoder = new LiteNetLibTransportConnection(); + clientListener.NetworkReceiveEvent += (_, reader, _, _) => + { + var frame = decoder.Receive(reader.GetRemainingBytes()); + if (LiteNetLibTransportConnection.DecodeSchema(frame) is CultNetErrorMessage error) Errors.Enqueue(error); + }; + _serverManager.Start(0); + _clientManager.Start(0); + _clientManager.Connect("127.0.0.1", _serverManager.LocalPort, "rwa-wire"); + if (!connected.Task.Wait(TimeSpan.FromSeconds(10))) Assert.Fail("the wire pair did not connect"); + var serverPeer = connected.Task.Result; + Peer = new CultNetServerPeer(serverPeer, new LiteNetLibTransportConnection(serverPeer)); + } + + public async Task NextErrorAsync() + { + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); + CultNetErrorMessage? error; + while (!Errors.TryDequeue(out error)) + { + if (DateTime.UtcNow > deadline) Assert.Fail("no reply arrived on the wire"); + await Task.Delay(5); + } + + return error; + } + + public void Dispose() + { + _clientManager.Stop(); + _serverManager.Stop(); + } + } + + [Test] + public async Task ThePeerReceivesTheTypedRefusalOnTheWire() + { + using var wire = new WirePair(); + var rig = Build(authority: null, peer: wire.Peer); + await Seed(rig, KeyA, "server"); + var logBefore = rig.LogLength; + + await rig.Put(rig.PutOf(KeyA, "peer")); + var putReply = await wire.NextErrorAsync(); + await rig.Delete(rig.DeleteOf(KeyA)); + var deleteReply = await wire.NextErrorAsync(); + await rig.Put(rig.PutOf(KeyB, "peer")); + var newKeyReply = await wire.NextErrorAsync(); + + foreach (var (reply, key) in new[] { (putReply, KeyA), (deleteReply, KeyA), (newKeyReply, KeyB) }) + { + Assert.That(reply.Code, Is.EqualTo("write_not_authorized")); + Assert.That(reply.Error, Does.StartWith("write_not_authorized: ")); + Assert.That(reply.Details!.Field, Is.EqualTo("recordKey")); + Assert.That(reply.Details.Value, Is.EqualTo(key.Value)); + } + + Assert.That(rig.TextOf(KeyA), Is.EqualTo("server")); + Assert.That(rig.TextOf(KeyB), Is.Null); + Assert.That(rig.LogLength, Is.EqualTo(logBefore)); + Assert.That(wire.Errors, Is.Empty, "one reply per refused write"); + } + + [Test] + public async Task AnAuthoritysRefusalReasonReachesThePeerAndAGrantSendsNothingBack() + { + using var wire = new WirePair(); + var rig = Build(new RecordingAuthority(request => request.RecordKey.Equals(KeyA) + ? CultNetRemoteWriteDecision.Grant() + : CultNetRemoteWriteDecision.Refuse("only a")), peer: wire.Peer); + + await rig.Put(rig.PutOf(KeyA, "peer")); + await rig.Put(rig.PutOf(KeyB, "peer")); + var reply = await wire.NextErrorAsync(); + + Assert.That(reply.Code, Is.EqualTo("write_not_authorized")); + Assert.That(reply.Error, Is.EqualTo("write_not_authorized: only a")); + Assert.That(reply.Details!.Value, Is.EqualTo(KeyB.Value)); + Assert.That(rig.TextOf(KeyA), Is.EqualTo("peer")); + await Task.Delay(100); + Assert.That(wire.Errors, Is.Empty, "the granted write was not answered with an error"); + } + + [Test] + public async Task ADeleteSentUnderASchemaAliasAsksTheAuthorityWithTheCanonicalId() + { + var authority = new RecordingAuthority(_ => CultNetRemoteWriteDecision.Grant()); + var rig = Build(authority); + await Seed(rig, KeyA, "server"); + var aliased = rig.DeleteOf(KeyA); + aliased.SchemaId = "tests.networking_note.v1"; + Assert.That(aliased.SchemaId, Is.Not.EqualTo(rig.SchemaId), "the fixture really sends an alias"); + + await rig.Delete(aliased); + + var request = authority.Requests.Single(); + Assert.That(request.Operation, Is.EqualTo(CultNetRemoteWriteOperation.Delete)); + Assert.That(request.SchemaId, Is.EqualTo(rig.SchemaId), "the authority sees the canonical id, not the alias as sent"); + Assert.That(rig.TextOf(KeyA), Is.Null, "the aliased delete was granted and applied"); + } + + [Test] + public async Task ANullDecisionRefusesPutAndDelete() + { + var rig = Build(new RecordingAuthority(_ => null!)); + await Seed(rig, KeyA, "server"); + + AssertAnswered(() => rig.Put(rig.PutOf(KeyA, "peer")), "a null decision is a refusal"); + AssertAnswered(() => rig.Delete(rig.DeleteOf(KeyA)), "a null decision is a refusal"); + AssertAnswered(() => rig.Put(rig.PutOf(KeyB, "peer")), "a null decision is a refusal"); + + Assert.That(rig.TextOf(KeyA), Is.EqualTo("server")); + Assert.That(rig.TextOf(KeyB), Is.Null); + Assert.That(rig.Logger.Warnings, Has.Some.Contains("raw put refused: the write was not authorized")); + Assert.That(rig.Logger.Warnings, Has.Some.Contains("raw delete refused: the write was not authorized")); + Assert.That(rig.Logger.Errors, Is.Empty, "a null decision is a refusal, not a fault"); + } + + private sealed class RecordingAuthority : ICultNetRemoteWriteAuthority + { + private readonly Func _decide; + public List Requests { get; } = new(); + public RecordingAuthority(Func decide) => _decide = decide; + + public CultNetRemoteWriteDecision Decide(CultNetRemoteWriteRequest request) + { + Requests.Add(request); + return _decide(request); + } + } + + private sealed class CountingForwarder : ICultNetShardWriteForwarder + { + public int Puts { get; private set; } + public int Deletes { get; private set; } + public Task ForwardPutAsync(CultNetShardDescriptor shard, CultNetDocumentPutRawMessage message) { Puts++; return Task.CompletedTask; } + public Task ForwardDeleteAsync(CultNetShardDescriptor shard, CultNetDocumentDeleteMessage message) { Deletes++; return Task.CompletedTask; } + } + + private sealed class CapturingLogger : GameCult.Logging.ILogger + { + public List Errors { get; } = new(); + public List Warnings { get; } = new(); + public void LogInfo(string message) { } + public void LogWarning(string message) => Warnings.Add(message); + public void LogError(string message) => Errors.Add(message); + public void LogDebug(string message) { } + } + } +} diff --git a/tests/GameCult.Networking.Tests/SoulP4cProbeTests.cs b/tests/GameCult.Networking.Tests/SoulP4cProbeTests.cs index a4796e7d..726dfea4 100644 --- a/tests/GameCult.Networking.Tests/SoulP4cProbeTests.cs +++ b/tests/GameCult.Networking.Tests/SoulP4cProbeTests.cs @@ -56,7 +56,7 @@ namespace GameCult.Networking.Tests using var server = new Server(cache, ServerSecurityOptions.Development()); var logger = new Logger(); server.Logger = logger; - using var databaseServer = new CultNetDatabaseServer(server, database); + using var databaseServer = new CultNetDatabaseServer(server, database, new CultNetDatabaseServerOptions { RemoteWriteAuthority = new TestGrantsEverything() }); var handler = (Func)typeof(CultNetDatabaseServer) .GetField("_putHandler", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)! .GetValue(databaseServer)!;