using System.Collections.Concurrent; using System.Diagnostics; using System.Globalization; using System.Net.WebSockets; using System.Reflection; using System.Text; using System.Text.Json; using System.Text.Json.Nodes; using Microsoft.Extensions.Options; using Nexus.Api.Observability; namespace Nexus.Api.Services; /// /// Maintains Nexus' single trusted backend connection to the OpenClaw Gateway. /// The connector implements the protocol-v4 challenge/connect handshake, keeps /// request/response correlation inside the backend, and never exposes Gateway /// credentials to the browser. /// public sealed class GatewayConnector : BackgroundService, IGatewayConnector { private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web); private readonly IConfiguration _configuration; private readonly ILogger _logger; private readonly GatewayConnectorOptions _options; private readonly IOpenClawDeviceIdentityStore _deviceIdentityStore; private readonly object _stateLock = new(); private readonly SemaphoreSlim _sendLock = new(1, 1); private readonly SemaphoreSlim _reconnectSignal = new(0, 1); private readonly ConcurrentDictionary> _pending = new(StringComparer.Ordinal); private readonly ConcurrentQueue _recentEvents = new(); private ClientWebSocket? _socket; private GatewayConnectionState _state = GatewayConnectionState.Initializing; private string? _gatewayVersion; private string? _requiredVersion; private DateTimeOffset? _lastConnectedAt; private DateTimeOffset? _lastEventAt; private int _reconnectAttempts; private int? _protocolVersion; private string? _statusMessage; private string? _deviceId; private bool _deviceTokenConfigured; private bool _pairingRequired; private string? _pairingRequestId; private HashSet _advertisedMethods = new(StringComparer.Ordinal); private HashSet _advertisedEvents = new(StringComparer.Ordinal); private HashSet _grantedScopes = new(StringComparer.Ordinal); private HashSet _requestedScopes = new(StringComparer.Ordinal); private Uri? _endpointOverride; private string? _tlsFingerprintOverride; private string? _transientBootstrapToken; public GatewayConnector( IConfiguration configuration, IOptions options, ILogger logger) : this( configuration, options, logger, new OpenClawDeviceIdentityStore( options, Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance)) { } public GatewayConnector( IConfiguration configuration, IOptions options, ILogger logger, IOpenClawDeviceIdentityStore deviceIdentityStore) { _configuration = configuration; _logger = logger; _options = options.Value; _deviceIdentityStore = deviceIdentityStore; _requestedScopes = _options.Scopes .Where(scope => !string.IsNullOrWhiteSpace(scope)) .Select(scope => scope.Trim()) .ToHashSet(StringComparer.Ordinal); var pinnedVersion = _configuration["Integrations:OpenClaw:RequiredVersion"]; _requiredVersion = string.IsNullOrWhiteSpace(pinnedVersion) ? OpenClawGatewayProtocol.DefaultRequiredGatewayVersion : pinnedVersion.Trim(); } public GatewayConnectionState ConnectionState { get { lock (_stateLock) return _state; } } public string? GatewayVersion { get { lock (_stateLock) return _gatewayVersion; } } public string? RequiredVersion { get { lock (_stateLock) return _requiredVersion; } } public DateTimeOffset? LastConnectedAt { get { lock (_stateLock) return _lastConnectedAt; } } public int ReconnectAttempts { get { lock (_stateLock) return _reconnectAttempts; } } public string? StatusMessage { get { lock (_stateLock) return _statusMessage; } } public string? DeviceId { get { lock (_stateLock) return _deviceId; } } public bool DeviceTokenConfigured { get { lock (_stateLock) return _deviceTokenConfigured; } } public bool PairingRequired { get { lock (_stateLock) return _pairingRequired; } } public string? PairingRequestId { get { lock (_stateLock) return _pairingRequestId; } } public int? ProtocolVersion { get { lock (_stateLock) return _protocolVersion; } } public IReadOnlySet AdvertisedMethods { get { lock (_stateLock) return new HashSet(_advertisedMethods, StringComparer.Ordinal); } } public IReadOnlySet AdvertisedEvents { get { lock (_stateLock) return new HashSet(_advertisedEvents, StringComparer.Ordinal); } } public IReadOnlySet GrantedScopes { get { lock (_stateLock) return new HashSet(_grantedScopes, StringComparer.Ordinal); } } public string? ActiveEndpoint { get { lock (_stateLock) return (_endpointOverride ?? BuildConfiguredWebSocketUri()).AbsoluteUri; } } public string? ActiveTlsFingerprint { get { lock (_stateLock) return _tlsFingerprintOverride ?? NormalizeFingerprint(_options.TlsFingerprint); } } public DateTimeOffset? LastEventAt { get { lock (_stateLock) return _lastEventAt; } } public bool Supports(string method) { if (string.IsNullOrWhiteSpace(method)) return false; lock (_stateLock) return _advertisedMethods.Contains(method); } public IReadOnlyList GetRecentEvents(int limit = 100) { var bounded = Math.Clamp(limit, 1, _options.EventBufferCapacity); return _recentEvents .ToArray() .Reverse() .Take(bounded) .Select(item => item with { Payload = item.Payload?.DeepClone() }) .ToList(); } public async Task RequestOperatorScopesAsync( IReadOnlyCollection scopes, CancellationToken cancellationToken = default) { ArgumentNullException.ThrowIfNull(scopes); var requested = scopes .Where(scope => !string.IsNullOrWhiteSpace(scope)) .Select(scope => scope.Trim()) .Distinct(StringComparer.Ordinal) .Order(StringComparer.Ordinal) .ToArray(); if (requested.Length == 0 || !requested.Contains("operator.read", StringComparer.Ordinal) || requested.Any(scope => scope is not ( "operator.read" or "operator.admin" or "operator.approvals"))) { throw new ArgumentException( "Only an explicit OpenClaw operator scope set containing operator.read can be requested.", nameof(scopes)); } ClientWebSocket? socket; lock (_stateLock) { _requestedScopes = requested.ToHashSet(StringComparer.Ordinal); if (requested.All(_grantedScopes.Contains)) return; socket = _socket; _pairingRequired = false; _pairingRequestId = null; _statusMessage = "A new OpenClaw operator scope set was requested; reconnecting for pairing."; } if (socket is not { State: WebSocketState.Open }) { SignalReconnect(); return; } await _sendLock.WaitAsync(cancellationToken); try { if (socket.State == WebSocketState.Open) { await socket.CloseOutputAsync( WebSocketCloseStatus.NormalClosure, "operator scope change", cancellationToken); } } catch (WebSocketException) { // The receive loop will reconnect and apply the requested scopes. } catch (ObjectDisposedException) { // The receive loop already observed the disconnect. } finally { _sendLock.Release(); } SignalReconnect(); } public async Task ConfigureEndpointAsync( string endpoint, string? tlsFingerprint, string? bootstrapToken, CancellationToken cancellationToken = default) { if (!Uri.TryCreate(endpoint?.Trim(), UriKind.Absolute, out var target) || target.Scheme is not ("ws" or "wss") || string.IsNullOrWhiteSpace(target.Host) || !string.IsNullOrEmpty(target.UserInfo) || !string.IsNullOrEmpty(target.Query) || !string.IsNullOrEmpty(target.Fragment)) { throw new ArgumentException("A normalized ws:// or wss:// OpenClaw endpoint is required.", nameof(endpoint)); } var normalizedFingerprint = NormalizeFingerprint(tlsFingerprint); if (!string.IsNullOrWhiteSpace(tlsFingerprint) && normalizedFingerprint.Length != 64) throw new ArgumentException("TLS fingerprint must be a SHA-256 hexadecimal value.", nameof(tlsFingerprint)); if (bootstrapToken is { Length: > 4096 } || bootstrapToken?.Any(char.IsControl) == true) throw new ArgumentException("Bootstrap token is invalid.", nameof(bootstrapToken)); ClientWebSocket? socket; lock (_stateLock) { var endpointChanged = _endpointOverride is null || !string.Equals(_endpointOverride.AbsoluteUri, target.AbsoluteUri, StringComparison.OrdinalIgnoreCase) || !string.Equals( _tlsFingerprintOverride ?? string.Empty, normalizedFingerprint, StringComparison.Ordinal); _endpointOverride = target; _tlsFingerprintOverride = normalizedFingerprint; if (endpointChanged) _transientBootstrapToken = null; if (!string.IsNullOrWhiteSpace(bootstrapToken)) _transientBootstrapToken = bootstrapToken; _requestedScopes = new HashSet(["operator.read"], StringComparer.Ordinal); _grantedScopes.Clear(); _pairingRequired = false; _pairingRequestId = null; _reconnectAttempts = 0; _state = GatewayConnectionState.Reconnecting; _statusMessage = $"Connecting to the validated OpenClaw endpoint {target.Host}."; socket = _socket; } if (socket is { State: WebSocketState.Open }) { await _sendLock.WaitAsync(cancellationToken); try { if (socket.State == WebSocketState.Open) { await socket.CloseOutputAsync( WebSocketCloseStatus.NormalClosure, "setup endpoint change", cancellationToken); } } catch (WebSocketException) { // The receive loop already observed the endpoint transition. } catch (ObjectDisposedException) { // The receive loop already disposed the socket. } finally { _sendLock.Release(); } } SignalReconnect(); } public async Task DisconnectAsync(CancellationToken cancellationToken = default) { ClientWebSocket? socket; lock (_stateLock) { socket = _socket; _deviceTokenConfigured = false; _pairingRequired = false; _pairingRequestId = null; _grantedScopes.Clear(); _statusMessage = "The adopted OpenClaw connection was detached locally."; } if (socket is not { State: WebSocketState.Open }) return; await _sendLock.WaitAsync(cancellationToken); try { if (socket.State == WebSocketState.Open) { await socket.CloseOutputAsync( WebSocketCloseStatus.NormalClosure, "connection detached", cancellationToken); } } catch (WebSocketException) { // The receive loop already observed the disconnect. } catch (ObjectDisposedException) { // The receive loop already disposed the socket. } finally { _sendLock.Release(); } } public async Task InvokeAsync( string method, object? parameters = null, TimeSpan? timeout = null, CancellationToken cancellationToken = default, OpenClawInvocationContext? invocationContext = null) { if (string.IsNullOrWhiteSpace(method)) throw new ArgumentException("Gateway method is required.", nameof(method)); using var activity = NexusTelemetry.ActivitySource.StartActivity( "nexus.openclaw.gateway.rpc", ActivityKind.Client); activity?.SetTag("rpc.system", "openclaw"); activity?.SetTag("rpc.method", method); var stopwatch = Stopwatch.StartNew(); var outcome = "error"; try { ClientWebSocket socket; lock (_stateLock) { if (_state != GatewayConnectionState.Connected || _socket is not { State: WebSocketState.Open } currentSocket) { throw new OpenClawGatewayRpcException( "GATEWAY_DISCONNECTED", "OpenClaw Gateway is not connected.", retryable: true); } if (_advertisedMethods.Count > 0 && !_advertisedMethods.Contains(method)) { throw new OpenClawGatewayRpcException( "METHOD_UNAVAILABLE", $"OpenClaw Gateway does not advertise '{method}'."); } socket = currentSocket; } var requestId = BuildGatewayRequestId(invocationContext?.CorrelationId); var completion = new TaskCompletionSource( TaskCreationOptions.RunContinuationsAsynchronously); if (!_pending.TryAdd(requestId, completion)) throw new InvalidOperationException("Could not allocate a Gateway request id."); var parametersNode = parameters switch { null => new JsonObject(), JsonNode node => node.DeepClone(), _ => JsonSerializer.SerializeToNode(parameters, JsonOptions) ?? new JsonObject() }; var request = OpenClawGatewayProtocol.BuildRpcRequest( requestId, method, parametersNode, invocationContext); try { await SendFrameAsync(socket, request, cancellationToken); var response = await completion.Task.WaitAsync( timeout ?? TimeSpan.FromSeconds(_options.RpcTimeoutSeconds), cancellationToken); outcome = "success"; return response; } catch (TimeoutException) { outcome = "timeout"; throw new OpenClawGatewayRpcException( "GATEWAY_TIMEOUT", $"OpenClaw Gateway method '{method}' timed out.", retryable: true); } finally { _pending.TryRemove(requestId, out _); } } catch (OperationCanceledException) { outcome = "cancelled"; throw; } catch (OpenClawGatewayRpcException) { if (outcome == "error") outcome = "gateway_error"; throw; } finally { stopwatch.Stop(); activity?.SetTag("rpc.outcome", outcome); NexusTelemetry.GatewayRpcDuration.Record( stopwatch.Elapsed.TotalMilliseconds, new KeyValuePair("outcome", outcome)); } } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { while (!stoppingToken.IsCancellationRequested) { try { await ConnectAndReceiveAsync(stoppingToken); } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { break; } catch (OpenClawGatewayRpcException exception) { CapturePairingFailure(exception); _logger.LogWarning( "OpenClaw Gateway handshake failed with {Code}: {Message}", exception.Code, exception.Message); IncrementReconnectAttempts(); SetState(GatewayConnectionState.Disconnected, BuildSafeErrorMessage(exception)); } catch (Exception exception) { _logger.LogWarning(exception, "OpenClaw Gateway connection failed"); IncrementReconnectAttempts(); SetState(GatewayConnectionState.Disconnected, exception.Message); } finally { lock (_stateLock) _socket = null; FailPending("OpenClaw Gateway connection closed."); } if (stoppingToken.IsCancellationRequested) break; if (ConnectionState == GatewayConnectionState.Failed) { try { await _reconnectSignal.WaitAsync(stoppingToken); continue; } catch (OperationCanceledException) { break; } } if (_options.ReconnectMaxAttempts > 0 && ReconnectAttempts >= _options.ReconnectMaxAttempts) { SetState( GatewayConnectionState.Failed, $"Reconnect limit reached after {ReconnectAttempts} attempts."); continue; } var delay = ComputeBackoff(); SetState(GatewayConnectionState.Reconnecting, $"Reconnect scheduled in {Math.Ceiling(delay.TotalSeconds)} seconds."); try { await _reconnectSignal.WaitAsync(delay, stoppingToken); } catch (OperationCanceledException) { break; } } } private async Task ConnectAndReceiveAsync(CancellationToken stoppingToken) { OpenClawGatewayProtocol.ValidateExternalClientIdentity(_options); using var socket = new ClientWebSocket(); socket.Options.KeepAliveInterval = TimeSpan.FromSeconds(30); socket.Options.UseDefaultCredentials = false; var endpoint = BuildWebSocketUri(); ConfigureTransportTrust(socket, endpoint); var gatewayBinding = BuildGatewayBinding(endpoint, GetActiveTlsFingerprint()); var requiresDeviceIdentity = OpenClawGatewayProtocol.RequiresDeviceIdentity(endpoint); SetState(GatewayConnectionState.Reconnecting, $"Connecting to {endpoint.Host}."); await socket.ConnectAsync(endpoint, stoppingToken); using var handshakeTimeout = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken); handshakeTimeout.CancelAfter(TimeSpan.FromSeconds(_options.HandshakeTimeoutSeconds)); var challenge = await ReceiveChallengeAsync(socket, handshakeTimeout.Token); if (string.IsNullOrWhiteSpace(challenge)) throw new OpenClawGatewayRpcException("INVALID_HANDSHAKE", "OpenClaw Gateway did not send connect.challenge."); var connectId = Guid.NewGuid().ToString("N"); var (configuredToken, configuredPassword) = ResolveCredentials(); string[] scopes; lock (_stateLock) scopes = _requestedScopes.Order(StringComparer.Ordinal).ToArray(); string? token = configuredToken; string? password = configuredPassword; string? deviceToken = null; OpenClawDeviceIdentity? identity = null; OpenClawGatewayDeviceProof? deviceProof = null; if (requiresDeviceIdentity) { identity = await _deviceIdentityStore.LoadOrCreateAsync(handshakeTimeout.Token); var storedToken = await _deviceIdentityStore.LoadTokenAsync( identity.DeviceId, "operator", gatewayBinding, handshakeTimeout.Token); if (storedToken is not null) { lock (_stateLock) _deviceTokenConfigured = true; if (string.IsNullOrWhiteSpace(configuredToken) && string.IsNullOrWhiteSpace(configuredPassword)) { deviceToken = storedToken.Token; token = storedToken.Token; password = null; } } var signedAt = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); var platform = GetPlatform(); var signaturePayload = OpenClawGatewayProtocol.BuildDeviceAuthPayloadV3( identity.DeviceId, _options.ClientId, _options.ClientMode, "operator", scopes, signedAt, token, challenge, platform, _options.DeviceFamily); deviceProof = new OpenClawGatewayDeviceProof( identity.DeviceId, identity.PublicKey, identity.Sign(signaturePayload), signedAt, challenge); lock (_stateLock) _deviceId = identity.DeviceId; } var connectFrame = OpenClawGatewayProtocol.BuildConnectRequest( connectId, _options, token, password, GetClientVersion(), GetPlatform(), CultureInfo.CurrentUICulture.Name, deviceProof, scopes, _options.DeviceFamily, deviceToken); await SendFrameAsync(socket, connectFrame, handshakeTimeout.Token); var hello = await ReceiveHelloAsync(socket, connectId, handshakeTimeout.Token); if (identity is not null && !string.IsNullOrWhiteSpace(hello.DeviceToken)) { await _deviceIdentityStore.StoreTokenAsync( identity.DeviceId, hello.Role ?? "operator", gatewayBinding, hello.DeviceToken, hello.Scopes, handshakeTimeout.Token); lock (_stateLock) { _deviceTokenConfigured = true; _transientBootstrapToken = null; } } if (hello.Protocol != _options.ProtocolVersion) { throw new OpenClawGatewayRpcException( "PROTOCOL_MISMATCH", $"Gateway negotiated protocol {hello.Protocol}; Nexus requires {_options.ProtocolVersion}."); } var (versionMatches, versionWarning) = EvaluateVersion(hello.ServerVersion); if (!versionMatches && _options.FailFastOnVersionMismatch) { SetState( GatewayConnectionState.Failed, versionWarning ?? "Gateway version mismatch."); await CloseBestEffortAsync(socket, "version mismatch"); return; } lock (_stateLock) { _socket = socket; _state = GatewayConnectionState.Connected; _gatewayVersion = hello.ServerVersion; _protocolVersion = hello.Protocol; _advertisedMethods = new HashSet(hello.Methods, StringComparer.Ordinal); _advertisedEvents = new HashSet(hello.Events, StringComparer.Ordinal); _grantedScopes = new HashSet(hello.Scopes, StringComparer.Ordinal); _lastConnectedAt = DateTimeOffset.UtcNow; _reconnectAttempts = 0; _statusMessage = versionWarning ?? $"Protocol v{hello.Protocol} ready; {_advertisedMethods.Count} methods advertised."; _pairingRequired = false; _pairingRequestId = null; } _logger.LogInformation( "OpenClaw Gateway connected using protocol {Protocol}; version {Version}; scopes {Scopes}", hello.Protocol, hello.ServerVersion ?? "unknown", string.Join(",", hello.Scopes)); await ReceiveLoopAsync(socket, stoppingToken); } private async Task ReceiveChallengeAsync(ClientWebSocket socket, CancellationToken cancellationToken) { for (var attempt = 0; attempt < 5; attempt++) { var frame = await ReceiveFrameAsync(socket, cancellationToken); if (frame is null) return null; if (OpenClawGatewayProtocol.IsConnectChallenge(frame, out var nonce)) return nonce; } return null; } private async Task ReceiveHelloAsync( ClientWebSocket socket, string connectId, CancellationToken cancellationToken) { for (var attempt = 0; attempt < 10; attempt++) { var frame = await ReceiveFrameAsync(socket, cancellationToken); if (frame is null) throw new OpenClawGatewayRpcException("INVALID_HANDSHAKE", "Gateway closed before hello-ok."); if (string.Equals(frame["type"]?.GetValue(), "res", StringComparison.Ordinal) && string.Equals(frame["id"]?.GetValue(), connectId, StringComparison.Ordinal)) { return OpenClawGatewayProtocol.ParseHello(frame, connectId); } } throw new OpenClawGatewayRpcException("INVALID_HANDSHAKE", "Gateway did not return hello-ok."); } private async Task ReceiveLoopAsync(ClientWebSocket socket, CancellationToken stoppingToken) { try { while (socket.State == WebSocketState.Open && !stoppingToken.IsCancellationRequested) { var frame = await ReceiveFrameAsync(socket, stoppingToken); if (frame is null) break; var type = frame["type"]?.GetValue(); if (string.Equals(type, "res", StringComparison.Ordinal)) { HandleResponse(frame); continue; } if (string.Equals(type, "event", StringComparison.Ordinal)) HandleEvent(frame); } } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { // Normal application shutdown. } catch (WebSocketException exception) { _logger.LogInformation("OpenClaw Gateway WebSocket closed: {Message}", exception.Message); } finally { await CloseBestEffortAsync(socket, "nexus connector shutdown"); if (!stoppingToken.IsCancellationRequested) { IncrementReconnectAttempts(); SetState(GatewayConnectionState.Disconnected, "Gateway connection closed."); } } } private void HandleResponse(JsonNode frame) { var requestId = frame["id"]?.GetValue(); if (string.IsNullOrWhiteSpace(requestId) || !_pending.TryRemove(requestId, out var completion)) return; if (frame["ok"]?.GetValue() == true) { completion.TrySetResult(frame["payload"]?.DeepClone()); return; } completion.TrySetException(OpenClawGatewayProtocol.CreateRpcException(frame["error"])); } private void HandleEvent(JsonNode frame) { var eventName = frame["event"]?.GetValue(); if (string.IsNullOrWhiteSpace(eventName)) return; var receivedAt = DateTimeOffset.UtcNow; var envelope = new GatewayEventEnvelope( eventName, frame["payload"]?.DeepClone(), TryGetLong(frame["seq"]), TryGetLong(frame["stateVersion"]), receivedAt); _recentEvents.Enqueue(envelope); while (_recentEvents.Count > _options.EventBufferCapacity) _recentEvents.TryDequeue(out _); lock (_stateLock) _lastEventAt = receivedAt; } private async Task SendFrameAsync( ClientWebSocket socket, JsonNode frame, CancellationToken cancellationToken) { var bytes = Encoding.UTF8.GetBytes(frame.ToJsonString(JsonOptions)); await _sendLock.WaitAsync(cancellationToken); try { await socket.SendAsync( new ArraySegment(bytes), WebSocketMessageType.Text, endOfMessage: true, cancellationToken); } finally { _sendLock.Release(); } } private async Task ReceiveFrameAsync( ClientWebSocket socket, CancellationToken cancellationToken) { var buffer = new byte[16 * 1024]; using var message = new MemoryStream(); while (true) { var result = await socket.ReceiveAsync(new ArraySegment(buffer), cancellationToken); if (result.MessageType == WebSocketMessageType.Close) return null; if (result.MessageType != WebSocketMessageType.Text) throw new OpenClawGatewayRpcException("UNSUPPORTED_FRAME", "Gateway returned a non-text frame."); await message.WriteAsync(buffer.AsMemory(0, result.Count), cancellationToken); if (message.Length > _options.MaxFrameBytes) throw new OpenClawGatewayRpcException("PAYLOAD_TOO_LARGE", "Gateway frame exceeded Nexus' configured receive limit."); if (result.EndOfMessage) break; } message.Position = 0; try { return await JsonNode.ParseAsync(message, cancellationToken: cancellationToken); } catch (JsonException exception) { throw new OpenClawGatewayRpcException("INVALID_FRAME", $"Gateway returned invalid JSON: {exception.Message}"); } } private Uri BuildWebSocketUri() { lock (_stateLock) return _endpointOverride ?? BuildConfiguredWebSocketUri(); } private Uri BuildConfiguredWebSocketUri() { var configuredBase = _configuration["Integrations:OpenClaw:BaseUrl"] ?? "http://127.0.0.1:18789"; var builder = new UriBuilder(configuredBase) { Scheme = configuredBase.StartsWith("https://", StringComparison.OrdinalIgnoreCase) ? "wss" : "ws" }; var configuredPath = string.IsNullOrWhiteSpace(_options.WebSocketPath) ? "/" : _options.WebSocketPath.Trim(); builder.Path = configuredPath.StartsWith('/') ? configuredPath : "/" + configuredPath; builder.Query = string.Empty; builder.Fragment = string.Empty; return builder.Uri; } private void ConfigureTransportTrust(ClientWebSocket socket, Uri endpoint) { if (string.Equals(endpoint.Scheme, "ws", StringComparison.OrdinalIgnoreCase)) { var trustedPlaintextHost = endpoint.IsLoopback || string.Equals(endpoint.Host, "localhost", StringComparison.OrdinalIgnoreCase) || _options.TrustedPlaintextHosts.Any(host => string.Equals(host?.Trim(), endpoint.Host, StringComparison.OrdinalIgnoreCase)); if (!trustedPlaintextHost) { throw new OpenClawGatewayRpcException( "INSECURE_GATEWAY_ENDPOINT", "Plaintext OpenClaw WebSockets are allowed only for loopback or explicitly trusted internal hosts."); } return; } if (!string.Equals(endpoint.Scheme, "wss", StringComparison.OrdinalIgnoreCase)) { throw new OpenClawGatewayRpcException( "INVALID_GATEWAY_ENDPOINT", "OpenClaw Gateway endpoint must use ws or wss."); } var expectedFingerprint = GetActiveTlsFingerprint(); if (endpoint.IsLoopback && string.IsNullOrWhiteSpace(expectedFingerprint)) return; if (string.IsNullOrWhiteSpace(expectedFingerprint)) { throw new OpenClawGatewayRpcException( "TLS_PIN_REQUIRED", "Remote OpenClaw WSS endpoints require an explicitly confirmed SHA-256 TLS fingerprint."); } socket.Options.RemoteCertificateValidationCallback = (_, certificate, _, errors) => { if (certificate is null || errors != System.Net.Security.SslPolicyErrors.None) return false; var actual = Convert.ToHexString(certificate.GetCertHash(System.Security.Cryptography.HashAlgorithmName.SHA256)); return string.Equals( NormalizeFingerprint(actual), expectedFingerprint, StringComparison.Ordinal); }; } private static string NormalizeFingerprint(string? value) => string.Concat((value ?? string.Empty) .Where(character => char.IsAsciiHexDigit(character))) .ToUpperInvariant(); internal static string BuildGatewayBinding(Uri endpoint, string? tlsFingerprint) { ArgumentNullException.ThrowIfNull(endpoint); var normalizedEndpoint = endpoint.GetComponents( UriComponents.SchemeAndServer | UriComponents.Path, UriFormat.UriEscaped).TrimEnd('/').ToLowerInvariant(); var material = $"{normalizedEndpoint}|{NormalizeFingerprint(tlsFingerprint)}"; return OpenClawInvocationContextFactory.Hash(material); } private (string? Token, string? Password) ResolveCredentials() { var password = _configuration["Integrations:OpenClaw:Password"]; var token = _configuration["Integrations:OpenClaw:Token"]; lock (_stateLock) { if (!string.IsNullOrWhiteSpace(_transientBootstrapToken)) token = _transientBootstrapToken; } return ( string.IsNullOrWhiteSpace(token) ? null : token, string.IsNullOrWhiteSpace(password) ? null : password); } private string GetActiveTlsFingerprint() { lock (_stateLock) return _tlsFingerprintOverride ?? NormalizeFingerprint(_options.TlsFingerprint); } private void SignalReconnect() { try { if (_reconnectSignal.CurrentCount == 0) _reconnectSignal.Release(); } catch (SemaphoreFullException) { // Another caller already woke the connector. } } private (bool Matches, string? Warning) EvaluateVersion(string? detectedVersion) { if (_requiredVersion is null) return (true, detectedVersion is null ? "Gateway version is unknown and unpinned." : null); if (detectedVersion is null) { var warning = $"Gateway did not report a version; Nexus requires '{_requiredVersion}'."; return (_options.FailFastOnMissingVersion ? false : true, warning); } if (!string.Equals(detectedVersion, _requiredVersion, StringComparison.OrdinalIgnoreCase)) return (false, $"Gateway version drift: detected '{detectedVersion}', required '{_requiredVersion}'."); return (true, null); } private TimeSpan ComputeBackoff() { var attempts = Math.Max(0, ReconnectAttempts); var backoffMs = Math.Min( _options.ReconnectInitialDelayMs * Math.Pow(2, attempts), _options.ReconnectMaxDelayMs); var jitter = backoffMs * 0.25 * (Random.Shared.NextDouble() * 2 - 1); return TimeSpan.FromMilliseconds(Math.Max(100, backoffMs + jitter)); } private void IncrementReconnectAttempts() { lock (_stateLock) _reconnectAttempts++; } private void SetState(GatewayConnectionState state, string? message = null) { lock (_stateLock) { _state = state; if (message is not null) _statusMessage = message; } } private void FailPending(string message) { foreach (var request in _pending.ToArray()) { if (_pending.TryRemove(request.Key, out var completion)) { completion.TrySetException(new OpenClawGatewayRpcException( "GATEWAY_DISCONNECTED", message, retryable: true)); } } } private static async Task CloseBestEffortAsync(ClientWebSocket socket, string description) { if (socket.State is not (WebSocketState.Open or WebSocketState.CloseReceived)) return; try { await socket.CloseOutputAsync( WebSocketCloseStatus.NormalClosure, description, CancellationToken.None); } catch { // Connection is already gone. } } private static string BuildSafeErrorMessage(OpenClawGatewayRpcException exception) { var recommendation = exception.Details?["recommendedNextStep"]?.GetValue(); return string.IsNullOrWhiteSpace(recommendation) ? $"{exception.Code}: {exception.Message}" : $"{exception.Code}: {exception.Message} ({recommendation})"; } private void CapturePairingFailure(OpenClawGatewayRpcException exception) { var pairingRequired = OpenClawGatewayProtocol.TryReadPairingRequest( exception, out var requestId); lock (_stateLock) { _pairingRequired = pairingRequired; _pairingRequestId = pairingRequired ? requestId : null; } } private static string BuildGatewayRequestId(string? correlationId) { var suffix = Guid.NewGuid().ToString("N"); if (string.IsNullOrWhiteSpace(correlationId)) return suffix; var safe = new string(correlationId .Where(character => char.IsAsciiLetterOrDigit(character) || character is '-' or '_' or '.') .Take(64) .ToArray()); return string.IsNullOrWhiteSpace(safe) ? suffix : $"{safe}.{suffix}"; } private static string GetClientVersion() { var version = Assembly.GetExecutingAssembly().GetName().Version; return version is null ? "0.1.0" : $"{version.Major}.{version.Minor}.{Math.Max(0, version.Build)}"; } private static string GetPlatform() { if (OperatingSystem.IsWindows()) return "windows"; if (OperatingSystem.IsMacOS()) return "macos"; if (OperatingSystem.IsLinux()) return "linux"; return "unknown"; } private static long? TryGetLong(JsonNode? node) { if (node is null) return null; try { return node.GetValueKind() switch { JsonValueKind.Number => node.GetValue(), JsonValueKind.String when long.TryParse(node.GetValue(), out var value) => value, _ => null }; } catch { return null; } } } public sealed class GatewayConnectorOptions { public const string SectionName = "GatewayConnector"; /// /// WebSocket path on the Gateway. OpenClaw serves the control plane at root by default. /// public string WebSocketPath { get; set; } = "/"; public int ProtocolVersion { get; set; } = OpenClawGatewayProtocol.CurrentProtocol; public int HandshakeTimeoutSeconds { get; set; } = 15; public int RpcTimeoutSeconds { get; set; } = 30; public int MaxFrameBytes { get; set; } = 4 * 1024 * 1024; public int EventBufferCapacity { get; set; } = 500; public string DeviceStatePath { get; set; } = string.Empty; public string OperationAuditPath { get; set; } = string.Empty; public string DeviceFamily { get; set; } = "server"; public string ClientId { get; set; } = "nexus"; public string ClientMode { get; set; } = "backend"; public string ClientDisplayName { get; set; } = "Nexus Mission Control"; public string ClientInstanceId { get; set; } = string.Empty; public bool ExternalClientIdentitySupported { get; set; } public bool AllowReservedInternalClientIdentity { get; set; } public string TlsFingerprint { get; set; } = string.Empty; public string[] TrustedPlaintextHosts { get; set; } = [ "openclaw-gateway", "host.docker.internal" ]; public int ReconnectInitialDelayMs { get; set; } = 1000; public int ReconnectMaxDelayMs { get; set; } = 300_000; public int ReconnectMaxAttempts { get; set; } public bool FailFastOnVersionMismatch { get; set; } = true; public bool FailFastOnMissingVersion { get; set; } public string[] Scopes { get; set; } = [ "operator.read" ]; public string[] Capabilities { get; set; } = [ "approvals", "exec-approvals", "session-scoped-events", "task-suggestions", "tool-events" ]; }