diff options
Diffstat (limited to 'Plugins')
| -rw-r--r-- | Plugins/CacheBroker/CacheBroker.csproj | 21 | ||||
| -rw-r--r-- | Plugins/CacheBroker/Endpoints/BrokerRegistrationEndpoint.cs | 54 | ||||
| -rw-r--r-- | Plugins/SessionCacheServer/Endpoints/ConnectEndpoint.cs | 195 | ||||
| -rw-r--r-- | Plugins/SessionCacheServer/ObjectCacheServer.csproj | 10 | ||||
| -rw-r--r-- | Plugins/SessionCacheServer/ObjectCacheServerEntry.cs | 100 | ||||
| -rw-r--r-- | Plugins/SessionProvider/SessionProvider - Backup (1).csproj | 60 | ||||
| -rw-r--r-- | Plugins/SessionProvider/SessionProvider - Backup.csproj | 56 | ||||
| -rw-r--r-- | Plugins/SessionProvider/SessionProvider.csproj | 20 | ||||
| -rw-r--r-- | Plugins/SessionProvider/null | 1 |
9 files changed, 392 insertions, 125 deletions
diff --git a/Plugins/CacheBroker/CacheBroker.csproj b/Plugins/CacheBroker/CacheBroker.csproj index 47a48cf..97a7e30 100644 --- a/Plugins/CacheBroker/CacheBroker.csproj +++ b/Plugins/CacheBroker/CacheBroker.csproj @@ -2,7 +2,10 @@ <PropertyGroup> <TargetFramework>net6.0</TargetFramework> - <Platforms>AnyCPU;x64</Platforms> + <Copyright>Copyright © 2022 Vaughn Nugent</Copyright> + <RootNamespace>VNLib.Plugins.Cache.Broker</RootNamespace> + <Authors>Vaughn Nugent</Authors> + <Version>1.0.1.2</Version> </PropertyGroup> <ItemGroup> @@ -26,10 +29,18 @@ <PropertyGroup> <!--Enable dynamic loading--> <EnableDynamicLoading>true</EnableDynamicLoading> - <Copyright>Copyright © 2022 Vaughn Nugent</Copyright> - <RootNamespace>VNLib.Plugins.Cache.Broker</RootNamespace> - <Authors>Vaughn Nugent</Authors> - <Version>1.0.0.1</Version> + <GenerateDocumentationFile>True</GenerateDocumentationFile> + <PackageProjectUrl>https://www.vaughnnugent.com/resources</PackageProjectUrl> + <AnalysisLevel>latest-all</AnalysisLevel> + + </PropertyGroup> + + <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|AnyCPU'"> + <Deterministic>False</Deterministic> + </PropertyGroup> + + <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'"> + <Deterministic>False</Deterministic> </PropertyGroup> diff --git a/Plugins/CacheBroker/Endpoints/BrokerRegistrationEndpoint.cs b/Plugins/CacheBroker/Endpoints/BrokerRegistrationEndpoint.cs index 2c454bf..340c47e 100644 --- a/Plugins/CacheBroker/Endpoints/BrokerRegistrationEndpoint.cs +++ b/Plugins/CacheBroker/Endpoints/BrokerRegistrationEndpoint.cs @@ -51,6 +51,7 @@ using VNLib.Plugins.Essentials.Extensions; using VNLib.Plugins.Extensions.Loading; using VNLib.Plugins.Extensions.Loading.Events; using VNLib.Net.Rest.Client; +using VaultSharp.V1.SystemBackend; #nullable enable @@ -101,7 +102,14 @@ namespace VNLib.Plugins.Cache.Broker.Endpoints private readonly Task<byte[]> ClientPubKey; private readonly Task<byte[]> BrokerPrivateKey; - protected override ProtectionSettings EndpointProtectionSettings { get; } + //Loosen up protection settings since this endpoint is not desinged for browsers or sessions + protected override ProtectionSettings EndpointProtectionSettings { get; } = new() + { + BrowsersOnly = false, + CrossSiteDenied = false, + SessionsRequired = false, + VerifySessionCors = false, + }; public BrokerRegistrationEndpoint(PluginBase plugin, IReadOnlyDictionary<string, JsonElement> config) { @@ -112,31 +120,22 @@ namespace VNLib.Plugins.Cache.Broker.Endpoints { _ = secret.Result ?? throw new InvalidOperationException("Broker private key not found in vault"); return Convert.FromBase64String(secret.Result); - }); + }, TaskScheduler.Default); CachePubKey = plugin.TryGetSecretAsync("cache_public_key").ContinueWith((Task<string?> secret) => { _ = secret.Result ?? throw new InvalidOperationException("Cache public key not found in vault"); return Convert.FromBase64String(secret.Result); - }); + }, TaskScheduler.Default); ClientPubKey = plugin.TryGetSecretAsync("client_public_key").ContinueWith((Task<string?> secret) => { _ = secret.Result ?? throw new InvalidOperationException("Client public key not found in vault"); return Convert.FromBase64String(secret.Result); - }); - + }, TaskScheduler.Default); InitPathAndLog(path, plugin.Log); - //Loosen up protection settings since this endpoint is not desinged for browsers or sessions - EndpointProtectionSettings = new() - { - SessionsRequired = false, - BrowsersOnly = false, - CrossSiteDenied = false, - }; - ListLock = new(); ActiveServers = new(); } @@ -149,7 +148,8 @@ namespace VNLib.Plugins.Cache.Broker.Endpoints //Verify with the client's pub key using (ECDsa alg = ECDsa.Create(DefaultCurve)) { - alg.ImportSubjectPublicKeyInfo(ClientPubKey.Result, out _); + ReadOnlyMemory<byte> client = await ClientPubKey; + alg.ImportSubjectPublicKeyInfo(client.Span, out _); //Verify with client public key if (!jwt.Verify(alg, in SignatureHashAlg)) { @@ -176,20 +176,14 @@ namespace VNLib.Plugins.Cache.Broker.Endpoints //Sign the jwt using the broker key using(ECDsa alg = ECDsa.Create(DefaultCurve)) { - alg.ImportPkcs8PrivateKey(BrokerPrivateKey.Result, out _); + ReadOnlyMemory<byte> brokerPrivate = await BrokerPrivateKey; + + alg.ImportPkcs8PrivateKey(brokerPrivate.Span, out _); response.Sign(alg, in SignatureHashAlg, 128); } - //Alloc output buffer - int bufSize = response.ByteSize * 2; - - using UnsafeMemoryHandle<char> charBuf = Memory.UnsafeAlloc<char>(bufSize, true); - - //compile jwt - ERRNO count = response.Compile(charBuf); - - entity.CloseResponse(HttpStatusCode.OK, ContentType.Text, charBuf.Span[..(int)count]); + entity.CloseResponse(HttpStatusCode.OK, ContentType.Text, response.DataBuffer); return VfReturnType.VirtualSkip; } catch (KeyNotFoundException) @@ -217,11 +211,13 @@ namespace VNLib.Plugins.Cache.Broker.Endpoints protected override async ValueTask<VfReturnType> PutAsync(HttpEntity entity) { //Parse jwt - using JsonWebToken? jwt = await entity.ParseFileAsAsync(ParseJwtAsync); + using JsonWebToken? jwt = await entity.ParseFileAsAsync(ParseJwtAsync) ?? throw new Exception(""); //Verify with the cache server's pub key using (ECDsa alg = ECDsa.Create(DefaultCurve)) { - alg.ImportSubjectPublicKeyInfo(CachePubKey.Result, out _); + ReadOnlyMemory<byte> cache = await CachePubKey; + + alg.ImportSubjectPublicKeyInfo(cache.Span, out _); //Verify the jwt if (!jwt.Verify(alg, in SignatureHashAlg)) { @@ -237,7 +233,7 @@ namespace VNLib.Plugins.Cache.Broker.Endpoints using JsonDocument requestBody = jwt.GetPayload(); //Get request keys - string? serverId = requestBody.RootElement.GetProperty("server_id").GetString(); + string? serverId = requestBody.RootElement.GetProperty("sub").GetString(); string? hostname = requestBody.RootElement.GetProperty("address").GetString(); string? token = requestBody.RootElement.GetProperty("token").GetString(); @@ -366,7 +362,9 @@ namespace VNLib.Plugins.Cache.Broker.Endpoints //Sign the jwt using the broker key using (ECDsa alg = ECDsa.Create(DefaultCurve)) { - alg.ImportPkcs8PrivateKey(BrokerPrivateKey.Result, out _); + ReadOnlyMemory<byte> broker = await BrokerPrivateKey; + + alg.ImportPkcs8PrivateKey(broker.Span, out _); //Sign with broker key jwt.Sign(alg, in SignatureHashAlg, 128); } diff --git a/Plugins/SessionCacheServer/Endpoints/ConnectEndpoint.cs b/Plugins/SessionCacheServer/Endpoints/ConnectEndpoint.cs index dc94825..fee1ea7 100644 --- a/Plugins/SessionCacheServer/Endpoints/ConnectEndpoint.cs +++ b/Plugins/SessionCacheServer/Endpoints/ConnectEndpoint.cs @@ -33,7 +33,10 @@ using System.Collections.Generic; using System.Security.Cryptography; using System.Collections.Concurrent; +using VNLib.Net.Http; +using VNLib.Hashing; using VNLib.Utils.Async; +using VNLib.Utils.Memory; using VNLib.Utils.Logging; using VNLib.Hashing.IdentityUtility; using VNLib.Net.Messaging.FBM; @@ -50,7 +53,6 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints { class ConnectEndpoint : ResourceEndpointBase { - const int MAX_RECV_BUF_SIZE = 1000 * 1024; const int MIN_RECV_BUF_SIZE = 8 * 1024; const int MAX_HEAD_BUF_SIZE = 2048; @@ -60,8 +62,9 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints const int MAX_EVENT_QUEUE_SIZE = 10000; const int MAX_RESPONSE_BUFFER_SIZE = 10 * 1024; - private static readonly Encoding FBMHeaderEncoding = Helpers.DefaultEncoding; + private static readonly TimeSpan AuthTokenExpiration = TimeSpan.FromSeconds(30); + private readonly string AudienceLocalServerId; private readonly ObjectCacheStore Store; private readonly PluginBase Pbase; @@ -71,7 +74,13 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints public uint ConnectedClients => _connectedClients; - protected override ProtectionSettings EndpointProtectionSettings { get; } + //Loosen up protection settings + protected override ProtectionSettings EndpointProtectionSettings { get; } = new() + { + BrowsersOnly = false, + SessionsRequired = false, + CrossSiteDenied = false + }; public ConnectEndpoint(string path, ObjectCacheStore store, PluginBase pbase) { @@ -79,18 +88,107 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints Store = store;//Load client public key to verify signed messages Pbase = pbase; - StatefulEventQueue = new(StringComparer.OrdinalIgnoreCase); + //Start the queue worker - _ = ChangeWorkerAsync().ConfigureAwait(false); + _ = pbase.DeferTask(() => ChangeWorkerAsync(pbase.UnloadToken), 10); + + AudienceLocalServerId = Guid.NewGuid().ToString("N"); + } + + /* + * Used as a client negotiation and verification request + * + * The token created during this request will be verified by the client + * and is already verified by this server, will be passed back + * via the authorization header during the websocket upgrade. + * + * This server must verify the authenticity of the returned token + * + * The tokens are very short lived as requests are intended to be made + * directly after verification + */ + + protected override async ValueTask<VfReturnType> GetAsync(HttpEntity entity) + { + //Parse jwt from authoriation + string? jwtAuth = entity.Server.Headers[HttpRequestHeader.Authorization]; + if (string.IsNullOrWhiteSpace(jwtAuth)) + { + entity.CloseResponse(HttpStatusCode.Unauthorized); + return VfReturnType.VirtualSkip; + } + + string? nodeId = null; + string? challenge = null; + + // Parse jwt + using (JsonWebToken jwt = JsonWebToken.Parse(jwtAuth)) + { + //Get the client public key + byte[] clientPub = await GetClientPubAsync(); + + //Init sig alg + using ECDsa verAlg = ECDsa.Create(FBMDataCacheExtensions.CacheCurve); + //Import client pub key + verAlg.ImportSubjectPublicKeyInfo(clientPub, out _); + + //verify signature for client + if (!jwt.Verify(verAlg, in FBMDataCacheExtensions.CacheJwtAlgorithm)) + { + Log.Information("Client signature verification failed"); + entity.CloseResponse(HttpStatusCode.Unauthorized); + return VfReturnType.VirtualSkip; + } + + //Recover json body + using JsonDocument doc = jwt.GetPayload(); + if (doc.RootElement.TryGetProperty("sub", out JsonElement servIdEl)) + { + nodeId = servIdEl.GetString(); + } + if (doc.RootElement.TryGetProperty("chl", out JsonElement challengeEl)) + { + challenge = challengeEl.GetString(); + } + } - //Loosen up protection settings - EndpointProtectionSettings = new() + Log.Debug("Received negotiation request from node {node}", nodeId); + + //Verified, now we can create an auth message with a short expiration + using JsonWebToken auth = new(); + auth.WriteHeader(FBMDataCacheExtensions.JwtMessageHeader.Span); + auth.InitPayloadClaim() + .AddClaim("aud", AudienceLocalServerId) + .AddClaim("exp", DateTimeOffset.UtcNow.Add(AuthTokenExpiration).ToUnixTimeSeconds()) + .AddClaim("nonce", RandomHash.GetRandomHex(8)) + .AddClaim("chl", challenge) + //Specify the server's node id if set + .AddClaim("sub", nodeId) + //Add negotiaion args + .AddClaim(FBMClient.REQ_HEAD_BUF_QUERY_ARG, MAX_HEAD_BUF_SIZE) + .AddClaim(FBMClient.REQ_RECV_BUF_QUERY_ARG, MAX_RECV_BUF_SIZE) + .AddClaim(FBMClient.REQ_MAX_MESS_QUERY_ARG, MAX_MESSAGE_SIZE) + .CommitClaims(); + + //Sign the auth message + ECDsa sigAlg = ECDsa.Create(FBMDataCacheExtensions.CacheCurve); + byte[] cachePrivate = await GetCachePrivateKeyAsync(); + try { - BrowsersOnly = false, - SessionsRequired = false, - CrossSiteDenied = false - }; + sigAlg.ImportPkcs8PrivateKey(cachePrivate, out _); + //sign jwt + auth.Sign(sigAlg, in FBMDataCacheExtensions.CacheJwtAlgorithm, 256); + } + finally + { + Memory.InitializeBlock(cachePrivate.AsSpan()); + sigAlg.Dispose(); + } + + //Close response + entity.CloseResponse(HttpStatusCode.OK, ContentType.Text, auth.DataBuffer); + return VfReturnType.VirtualSkip; } private async Task<byte[]> GetClientPubAsync() @@ -99,15 +197,31 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints return Convert.FromBase64String(brokerPubKey); } + private async Task<byte[]> GetCachePubAsync() + { + string? brokerPubKey = await Pbase.TryGetSecretAsync("cache_public_key") ?? throw new KeyNotFoundException("Missing required secret : client_public_key"); + + return Convert.FromBase64String(brokerPubKey); + } + private async Task<byte[]> GetCachePrivateKeyAsync() + { + string? cachePrivate = await Pbase.TryGetSecretAsync("cache_private_key") ?? throw new KeyNotFoundException("Missing required secret : client_public_key"); + + byte[] data = Convert.FromBase64String(cachePrivate); + + Memory.UnsafeZeroMemory<char>(cachePrivate); + + return data; + } - private async Task ChangeWorkerAsync() + private async Task ChangeWorkerAsync(CancellationToken cancellation) { try { //Listen for changes while (true) { - ChangeEvent ev = await Store.EventQueue.DequeueAsync(Pbase.UnloadToken); + ChangeEvent ev = await Store.EventQueue.DequeueAsync(cancellation); //Add event to queues foreach (AsyncQueue<ChangeEvent> queue in StatefulEventQueue.Values) { @@ -139,45 +253,70 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints { try { - //Parse jwt from authoriation + //Parse jwt from authorization string? jwtAuth = entity.Server.Headers[HttpRequestHeader.Authorization]; if (string.IsNullOrWhiteSpace(jwtAuth)) { entity.CloseResponse(HttpStatusCode.Unauthorized); return VfReturnType.VirtualSkip; } + string? nodeId = null; //Parse jwt using (JsonWebToken jwt = JsonWebToken.Parse(jwtAuth)) { - //Get the client public key - byte[] clientPub = await GetClientPubAsync(); - - //Init sig alg - using ECDsa sigAlg = ECDsa.Create(FBMDataCacheExtensions.CacheCurve); - //Import client pub key - sigAlg.ImportSubjectPublicKeyInfo(clientPub, out _); - //verify signature for client - if (!jwt.Verify(sigAlg, FBMDataCacheExtensions.CacheJwtAlgorithm)) + //Init sig alg, we will verify that the token was signed by this server + using (ECDsa sigAlg = ECDsa.Create(FBMDataCacheExtensions.CacheCurve)) { - entity.CloseResponse(HttpStatusCode.Unauthorized); - return VfReturnType.VirtualSkip; + //Get the cache public key + byte[] cachePub = await GetCachePubAsync(); + + sigAlg.ImportSubjectPublicKeyInfo(cachePub, out _); + + //verify signature against the cache public key, since this server have signed it + if (!jwt.Verify(sigAlg, in FBMDataCacheExtensions.CacheJwtAlgorithm)) + { + entity.CloseResponse(HttpStatusCode.Unauthorized); + return VfReturnType.VirtualSkip; + } } + //Recover json body using JsonDocument doc = jwt.GetPayload(); - if (doc.RootElement.TryGetProperty("server_id", out JsonElement servIdEl)) + + //Verify audience, expiration + + if (!doc.RootElement.TryGetProperty("aud", out JsonElement audEl) + || AudienceLocalServerId.Equals(audEl.GetString(), StringComparison.OrdinalIgnoreCase)) + { + entity.CloseResponse(HttpStatusCode.Unauthorized); + return VfReturnType.VirtualSkip; + } + + if (!doc.RootElement.TryGetProperty("exp", out JsonElement expEl) + || DateTimeOffset.FromUnixTimeSeconds(expEl.GetInt64()) < DateTimeOffset.UtcNow) + { + entity.CloseResponse(HttpStatusCode.Unauthorized); + return VfReturnType.VirtualSkip; + } + + //The node id is optional and stored in the 'sub' field + if (doc.RootElement.TryGetProperty("sub", out JsonElement servIdEl)) { nodeId = servIdEl.GetString(); } } + //Get query config suggestions from the client string recvBufCmd = entity.QueryArgs[FBMClient.REQ_RECV_BUF_QUERY_ARG]; string maxHeaderCharCmd = entity.QueryArgs[FBMClient.REQ_HEAD_BUF_QUERY_ARG]; string maxMessageSizeCmd = entity.QueryArgs[FBMClient.REQ_MAX_MESS_QUERY_ARG]; + //Parse recv buffer size int recvBufSize = int.TryParse(recvBufCmd, out int rbs) ? rbs : MIN_RECV_BUF_SIZE; int maxHeadBufSize = int.TryParse(maxHeaderCharCmd, out int hbs) ? hbs : MIN_HEAD_BUF_SIZE; int maxMessageSize = int.TryParse(maxMessageSizeCmd, out int mxs) ? mxs : MIN_MESSAGE_SIZE; + AsyncQueue<ChangeEvent>? nodeQueue = null; //The connection may be a caching server node, so get its node-id if (!string.IsNullOrWhiteSpace(nodeId)) @@ -212,7 +351,9 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints MaxResponseBufferSize = Math.Min(maxMessageSize, MAX_RESPONSE_BUFFER_SIZE), SyncQueue = nodeQueue }; + Log.Debug("Client recv buffer suggestion {recv}, header buffer size {head}, response buffer size {r}", recvBufCmd, maxHeaderCharCmd, state.MaxResponseBufferSize); + //Accept socket and pass state object entity.AcceptWebSocket(WebsocketAcceptedAsync, state); return VfReturnType.VirtualSkip; @@ -239,7 +380,7 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints RecvBufferSize = state.RecvBufferSize, ResponseBufferSize = state.MaxResponseBufferSize, MaxHeaderBufferSize = state.MaxHeaderBufferSize, - HeaderEncoding = FBMHeaderEncoding, + HeaderEncoding = Helpers.DefaultEncoding, }; //Listen for requests diff --git a/Plugins/SessionCacheServer/ObjectCacheServer.csproj b/Plugins/SessionCacheServer/ObjectCacheServer.csproj index 5598acc..01b33f3 100644 --- a/Plugins/SessionCacheServer/ObjectCacheServer.csproj +++ b/Plugins/SessionCacheServer/ObjectCacheServer.csproj @@ -2,17 +2,21 @@ <PropertyGroup> <TargetFramework>net6.0</TargetFramework> + <Nullable>enable</Nullable> <Authors>Vaughn Nugent</Authors> - <Version>1.0.0.1</Version> + <Version>1.0.1.1</Version> <RootNamespace>VNLib.Plugins.Essentials.Sessions.Server</RootNamespace> <Copyright>Copyright © 2022 Vaughn Nugent</Copyright> - <Platforms>AnyCPU;x64</Platforms> + </PropertyGroup> <!-- Resolve nuget dll files and store them in the output dir --> <PropertyGroup> <EnableDynamicLoading>true</EnableDynamicLoading> - <Nullable>enable</Nullable> + <GenerateDocumentationFile>False</GenerateDocumentationFile> + <PackageProjectUrl>https://www.vaughnnugent.com/resources</PackageProjectUrl> + <AnalysisLevel>latest-all</AnalysisLevel> + </PropertyGroup> <ItemGroup> <Compile Remove="liveplugin2\**" /> diff --git a/Plugins/SessionCacheServer/ObjectCacheServerEntry.cs b/Plugins/SessionCacheServer/ObjectCacheServerEntry.cs index 6c969ba..17d8ba5 100644 --- a/Plugins/SessionCacheServer/ObjectCacheServerEntry.cs +++ b/Plugins/SessionCacheServer/ObjectCacheServerEntry.cs @@ -29,7 +29,6 @@ using System.Linq; using System.Net.Http; using System.Text.Json; using System.Threading; -using System.Diagnostics; using System.Net.Sockets; using System.Threading.Tasks; using System.Collections.Generic; @@ -50,6 +49,7 @@ using VNLib.Plugins.Cache.Broker.Endpoints; using VNLib.Plugins.Extensions.Loading; using VNLib.Plugins.Extensions.Loading.Routing; using VNLib.Plugins.Essentials.Sessions.Server.Endpoints; +using VNLib.Utils.Memory.Caching; namespace VNLib.Plugins.Essentials.Sessions.Server { @@ -72,16 +72,16 @@ namespace VNLib.Plugins.Essentials.Sessions.Server string swapDir = PluginConfig.GetProperty("swap_dir").GetString() ?? throw new KeyNotFoundException("Missing required key 'swap_dir' for config"); int cacheSize = PluginConfig.GetProperty("max_cache").GetInt32(); string connectPath = PluginConfig.GetProperty("connect_path").GetString() ?? throw new KeyNotFoundException("Missing required element 'connect_path' for config 'cluster'"); - TimeSpan cleanupInterval = PluginConfig.GetProperty("cleanup_interval_sec").GetTimeSpan(TimeParseType.Seconds); - TimeSpan validFor = PluginConfig.GetProperty("valid_for_sec").GetTimeSpan(TimeParseType.Seconds); + //TimeSpan cleanupInterval = PluginConfig.GetProperty("cleanup_interval_sec").GetTimeSpan(TimeParseType.Seconds); + //TimeSpan validFor = PluginConfig.GetProperty("valid_for_sec").GetTimeSpan(TimeParseType.Seconds); int maxMessageSize = PluginConfig.GetProperty("max_blob_size").GetInt32(); - TimeSpan initialCleanupDelay = TimeSpan.FromSeconds(2); //Init dir DirectoryInfo dir = new(swapDir); dir.Create(); //Init cache listener, single threaded reader ObjectCacheStore CacheListener = new(dir, cacheSize, Log, CacheHeap, true); + //Init connect endpoint { //Init connect endpoint @@ -96,19 +96,20 @@ namespace VNLib.Plugins.Essentials.Sessions.Server //Route the broker endpoint BrokerHeartBeat brokerEp = new(() => BrokerHeartBeatToken!, mre, new Uri(brokerAddress), this); Route(brokerEp); + //start registration - _ = RegisterServerAsync(mre) - .ConfigureAwait(false); + _ = this.DeferTask(() => RegisterServerAsync(mre), 200); } + //Setup cluster worker { //Get pre-configured fbm client config for caching FBMClientConfig conf = FBMDataCacheExtensions.GetDefaultConfig(CacheHeap, maxMessageSize, this.IsDebug() ? Log : null); //Start Client runner - _ = RunClientAsync(CacheListener, new Uri(brokerAddress), conf) - .ConfigureAwait(false); + _ = this.DeferTask(() => RunClientAsync(CacheListener, new Uri(brokerAddress), conf), 300); } + //Load a cache broker to the current server if the config is defined { if(this.HasConfigForType<BrokerRegistrationEndpoint>()) @@ -116,13 +117,10 @@ namespace VNLib.Plugins.Essentials.Sessions.Server this.Route<BrokerRegistrationEndpoint>(); } } - //Init timer and fire immediatly to cleanup - Timer CleanupTimer = new((object? state) => OnCleanupElapsed(state, validFor), CacheListener, initialCleanupDelay, cleanupInterval); - + void Cleanup() { CacheHeap.Dispose(); - CleanupTimer.Dispose(); CacheListener.Dispose(); } @@ -148,24 +146,6 @@ namespace VNLib.Plugins.Essentials.Sessions.Server Log.Information("Plugin unloaded"); } - private void OnCleanupElapsed(object? state, TimeSpan validFor) - { - try - { - ObjectCacheStore listener = state as ObjectCacheStore; - Stopwatch sw = new(); - sw.Start(); - //Cleanup - //await listener.CleanupExpiredAsync(validFor); - sw.Stop(); - Log.Debug("Expired cache records cleaned in {ms} ms", sw.Elapsed.TotalMilliseconds); - } - catch (Exception ex) - { - Log.Error(ex); - } - } - #region Registration private async Task RegisterServerAsync(ManualResetEvent keepaliveWait) @@ -175,12 +155,12 @@ namespace VNLib.Plugins.Essentials.Sessions.Server //Get the broker config element IReadOnlyDictionary<string, JsonElement> clusterConfig = this.GetConfig("cluster"); - Uri brokerAddress = new(clusterConfig["broker_address"].GetString()); + //Server id is just dns name for now string serverId = Dns.GetHostName(); int heartBeatDelayMs = clusterConfig["heartbeat_timeout_sec"].GetInt32() * 1000; - string connectPath = PluginConfig.GetProperty("connect_path").GetString(); + string? connectPath = PluginConfig.GetProperty("connect_path").GetString(); //Get the port of the primary webserver int port; @@ -195,14 +175,27 @@ namespace VNLib.Plugins.Essentials.Sessions.Server //If a certificate is specified, tls is enabled on the port usingTls = firstHost.TryGetProperty("cert", out _); } - - //Try to get the cache private key - string base64Priv = await this.TryGetSecretAsync("cache_private_key") ?? throw new KeyNotFoundException("Failed to load the cache private key"); - - byte[] privKey = Convert.FromBase64String(base64Priv); - //Init url builder for payload, see if tls is enabled - Uri connectAddress = new UriBuilder(usingTls ? Uri.UriSchemeHttps : Uri.UriSchemeHttp, Dns.GetHostName(), port, connectPath).Uri; + using BrokerRegistrationRequest request = new(); + { + string addr = clusterConfig["broker_address"].GetString() ?? throw new KeyNotFoundException("Missing required key 'broker_address' for config 'cluster'"); + + //Try to get the cache private key + string base64Priv = await this.TryGetSecretAsync("cache_private_key") ?? throw new KeyNotFoundException("Failed to load the cache private key"); + + ReadOnlyMemory<byte> privKey = Convert.FromBase64String(base64Priv); + + //Init url builder for payload, see if tls is enabled + Uri connectAddress = new UriBuilder(usingTls ? Uri.UriSchemeHttps : Uri.UriSchemeHttp, Dns.GetHostName(), port, connectPath).Uri; + + request.WithBroker(new(addr)) + .WithRegistrationAddress(connectAddress.ToString()) + .WithNodeId(serverId) + .WithPrivateKey(privKey.Span); + //Wipe memory + Memory.UnsafeZeroMemory<char>(base64Priv); + Memory.UnsafeZeroMemory(privKey); + } while (true) { @@ -210,11 +203,13 @@ namespace VNLib.Plugins.Essentials.Sessions.Server { //Gen a random reg token before registering BrokerHeartBeatToken = RandomHash.GetRandomHex(32); + //Assign new hb token + request.WithHeartbeatToken(BrokerHeartBeatToken); - Log.Information("Registering with cache broker {addr}, with node-id {id}", brokerAddress, serverId); + Log.Information("Registering with cache broker {addr}, with node-id {id}", request.BrokerAddress, serverId); //Register with the broker - await FBMDataCacheExtensions.ResgisterWithBrokerAsync(brokerAddress, privKey, connectAddress.ToString(), serverId, BrokerHeartBeatToken); + await FBMDataCacheExtensions.ResgisterWithBrokerAsync(request); Log.Debug("Successfully registered with cache broker"); @@ -357,15 +352,12 @@ namespace VNLib.Plugins.Essentials.Sessions.Server await Task.Delay(noServerDelay, UnloadToken); continue; } + //Select servers that are not the current server and are not already being monitored - IEnumerable<ActiveServer> serversToConnectTo = from s in - (from ss in servers - where ss.ServerId != nodeId - select ss) - where !ActiveServers.ContainsKey(s.ServerId) - select s; + IEnumerable<ActiveServer> serversToConnectTo = servers.Where(s => s.ServerId != nodeId) + .Where(s => !ActiveServers.ContainsKey(s.ServerId!)); //Connect to servers - foreach(ActiveServer server in serversToConnectTo) + foreach (ActiveServer server in serversToConnectTo) { _ = RunSyncTaskAsync(server, ActiveServers, cacheStore, clientConf, clientPrivKey, nodeId) .ConfigureAwait(false); @@ -439,10 +431,16 @@ namespace VNLib.Plugins.Essentials.Sessions.Server client.ReturnRequest(modRequest); } } - - string challenge = RandomHash.GetRandomBase64(24); + + //Configure cache + client.GetCacheConfiguration() + .ImportVerificationKey(privateKey.Span) + .ImportVerificationKey(null) + .WithNodeId(nodeId) //set nodeid since were listening for changes + .WithTls(false); + //Connect to the server - await client.ConnectAsync(server.HostName, privateKey, challenge, nodeId, false, UnloadToken); + await client.ConnectToCacheAsync(server, UnloadToken); //Wroker task callback method async Task BgWorkerAsync() diff --git a/Plugins/SessionProvider/SessionProvider - Backup (1).csproj b/Plugins/SessionProvider/SessionProvider - Backup (1).csproj new file mode 100644 index 0000000..7eb7022 --- /dev/null +++ b/Plugins/SessionProvider/SessionProvider - Backup (1).csproj @@ -0,0 +1,60 @@ +<Project Sdk="Microsoft.NET.Sdk"> + + <PropertyGroup> + <TargetFramework>net6.0</TargetFramework> + <RootNamespace>VNLib.Plugins.Essentials.Sessions</RootNamespace> + <AssemblyName>SessionProvider</AssemblyName> + <PackageId>SessionProvider</PackageId> + <Authors>Vaughn Nugent</Authors> + <Product>SessionProvider</Product> + <Copyright>Copyright © 2022 Vaughn Nugent</Copyright> + <Version>1.0.3.1</Version> + <PackageProjectUrl>www.vaughnnugent.com/resources</PackageProjectUrl> + <Platforms>AnyCPU;x64</Platforms> + </PropertyGroup> + + <ItemGroup> + <PackageReference Include="ErrorProne.NET.CoreAnalyzers" Version="0.1.2"> + <PrivateAssets>all</PrivateAssets> + <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets> + </PackageReference> + <PackageReference Include="ErrorProne.NET.Structs" Version="0.1.2"> + <PrivateAssets>all</PrivateAssets> + <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets> + </PackageReference> + </ItemGroup> + + <!-- Resolve nuget dll files and store them in the output dir --> + <PropertyGroup> + <EnableDynamicLoading>true</EnableDynamicLoading> + <AssemblyVersion>1.0.2.1</AssemblyVersion> + <FileVersion>1.0.2.1</FileVersion> + <Nullable>enable</Nullable> + </PropertyGroup> + <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'"> + <Deterministic>False</Deterministic> + </PropertyGroup> + <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x64'"> + <Deterministic>False</Deterministic> + </PropertyGroup> + <ItemGroup> + <ProjectReference Include="..\..\..\..\VNLib\Utils\src\VNLib.Utils.csproj" /> + <ProjectReference Include="..\..\..\Extensions\VNLib.Plugins.Extensions.Loading\VNLib.Plugins.Extensions.Loading.csproj" /> + <ProjectReference Include="..\..\Libs\VNLib.Plugins.Essentials.Sessions.Runtime\VNLib.Plugins.Essentials.Sessions.Runtime.csproj" /> + </ItemGroup> + + <ItemGroup> + <None Update="SessionProvider.json"> + <CopyToOutputDirectory>Always</CopyToOutputDirectory> + </None> + </ItemGroup> + + <Target Name="PostBuild" AfterTargets="PostBuildEvent"> + <Exec Command="start xcopy "$(TargetDir)" "F:\Programming\Web Plugins\DevPlugins\$(TargetName)" /E /Y /R" /> + </Target> + + <Target Name="PreBuild" BeforeTargets="PreBuildEvent"> + <Exec Command="erase "F:\Programming\Web Plugins\DevPlugins\$(TargetName)" > nul" /> + </Target> + +</Project> diff --git a/Plugins/SessionProvider/SessionProvider - Backup.csproj b/Plugins/SessionProvider/SessionProvider - Backup.csproj new file mode 100644 index 0000000..d20a524 --- /dev/null +++ b/Plugins/SessionProvider/SessionProvider - Backup.csproj @@ -0,0 +1,56 @@ +<Project Sdk="Microsoft.NET.Sdk"> + + <PropertyGroup> + <TargetFramework>net6.0</TargetFramework> + <RootNamespace>VNLib.Plugins.Essentials.Sessions</RootNamespace> + <AssemblyName>SessionProvider</AssemblyName> + <PackageId>SessionProvider</PackageId> + <Authors>Vaughn Nugent</Authors> + <Product>SessionProvider</Product> + <Copyright>Copyright © 2022 Vaughn Nugent</Copyright> + <Version>1.0.3.1</Version> + <PackageProjectUrl>www.vaughnnugent.com/resources</PackageProjectUrl> + <Platforms>AnyCPU;x64</Platforms> + </PropertyGroup> + + <ItemGroup> + <PackageReference Include="ErrorProne.NET.CoreAnalyzers" Version="0.1.2"> + <PrivateAssets>all</PrivateAssets> + <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets> + </PackageReference> + <PackageReference Include="ErrorProne.NET.Structs" Version="0.1.2"> + <PrivateAssets>all</PrivateAssets> + <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets> + </PackageReference> + </ItemGroup> + + <!-- Resolve nuget dll files and store them in the output dir --> + <PropertyGroup> + <EnableDynamicLoading>true</EnableDynamicLoading> + <AssemblyVersion>1.0.2.1</AssemblyVersion> + <FileVersion>1.0.2.1</FileVersion> + <Nullable>enable</Nullable> + </PropertyGroup> + <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'"> + <Deterministic>False</Deterministic> + </PropertyGroup> + <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x64'"> + <Deterministic>False</Deterministic> + </PropertyGroup> + <ItemGroup> + <ProjectReference Include="..\..\..\..\VNLib\Utils\src\VNLib.Utils.csproj" /> + <ProjectReference Include="..\..\..\Extensions\VNLib.Plugins.Extensions.Loading\VNLib.Plugins.Extensions.Loading.csproj" /> + <ProjectReference Include="..\..\Libs\VNLib.Plugins.Essentials.Sessions.Runtime\VNLib.Plugins.Essentials.Sessions.Runtime.csproj" /> + </ItemGroup> + + <ItemGroup> + <None Update="SessionProvider.json"> + <CopyToOutputDirectory>Always</CopyToOutputDirectory> + </None> + </ItemGroup> + + <Target Name="PostBuild" AfterTargets="PostBuildEvent"> + <Exec Command="start xcopy "$(TargetDir)" "F:\Programming\Web Plugins\DevPlugins\$(TargetName)" /E /Y /R" /> + </Target> + +</Project> diff --git a/Plugins/SessionProvider/SessionProvider.csproj b/Plugins/SessionProvider/SessionProvider.csproj index d20a524..4ed7dc1 100644 --- a/Plugins/SessionProvider/SessionProvider.csproj +++ b/Plugins/SessionProvider/SessionProvider.csproj @@ -2,6 +2,7 @@ <PropertyGroup> <TargetFramework>net6.0</TargetFramework> + <Nullable>enable</Nullable> <RootNamespace>VNLib.Plugins.Essentials.Sessions</RootNamespace> <AssemblyName>SessionProvider</AssemblyName> <PackageId>SessionProvider</PackageId> @@ -9,8 +10,7 @@ <Product>SessionProvider</Product> <Copyright>Copyright © 2022 Vaughn Nugent</Copyright> <Version>1.0.3.1</Version> - <PackageProjectUrl>www.vaughnnugent.com/resources</PackageProjectUrl> - <Platforms>AnyCPU;x64</Platforms> + <PackageProjectUrl>https://www.vaughnnugent.com/resources</PackageProjectUrl> </PropertyGroup> <ItemGroup> @@ -27,16 +27,10 @@ <!-- Resolve nuget dll files and store them in the output dir --> <PropertyGroup> <EnableDynamicLoading>true</EnableDynamicLoading> - <AssemblyVersion>1.0.2.1</AssemblyVersion> - <FileVersion>1.0.2.1</FileVersion> - <Nullable>enable</Nullable> - </PropertyGroup> - <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'"> - <Deterministic>False</Deterministic> - </PropertyGroup> - <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x64'"> - <Deterministic>False</Deterministic> + <GenerateDocumentationFile>True</GenerateDocumentationFile> + <AnalysisLevel>latest-all</AnalysisLevel> </PropertyGroup> + <ItemGroup> <ProjectReference Include="..\..\..\..\VNLib\Utils\src\VNLib.Utils.csproj" /> <ProjectReference Include="..\..\..\Extensions\VNLib.Plugins.Extensions.Loading\VNLib.Plugins.Extensions.Loading.csproj" /> @@ -53,4 +47,8 @@ <Exec Command="start xcopy "$(TargetDir)" "F:\Programming\Web Plugins\DevPlugins\$(TargetName)" /E /Y /R" /> </Target> + <Target Name="PreBuild" BeforeTargets="PreBuildEvent"> + <Exec Command="erase "F:\Programming\Web Plugins\DevPlugins\$(TargetName)" /q > nul" /> + </Target> + </Project> diff --git a/Plugins/SessionProvider/null b/Plugins/SessionProvider/null new file mode 100644 index 0000000..48b2100 --- /dev/null +++ b/Plugins/SessionProvider/null @@ -0,0 +1 @@ +F:\Programming\Web Plugins\DevPlugins\SessionProvider\*, Are you sure (Y/N)? |