using System;

using System.Buffers;

using System.Buffers.Binary;

using System.Collections.Generic;

using System.Data;

using System.Diagnostics;

using System.Diagnostics.CodeAnalysis;

using System.IO;

using System.Net;

using System.Net.Security;

using System.Net.Sockets;

using System.Runtime.CompilerServices;

using System.Runtime.InteropServices;

using System.Runtime.ExceptionServices;

using System.Security.Authentication;

using System.Security.Cryptography.X509Certificates;

using System.Text;

using System.Threading;

using System.Threading.Channels;

using System.Threading.Tasks;

using Npgsql.BackendMessages;

using Npgsql.Util;

using Microsoft.Extensions.Logging;

using Npgsql.Properties;

using static Npgsql.Util.Statics;

namespace Npgsql.Internal;

/// <summary>

/// Represents a connection to a PostgreSQL backend. Unlike NpgsqlConnection objects, which are

/// exposed to users, connectors are internal to Npgsql and are recycled by the connection pool.

/// </summary>

[Experimental(NpgsqlDiagnostics.ConvertersExperimental)]

public sealed partial class NpgsqlConnector

{

#region Fields and Properties

/// <summary>

/// The physical connection socket to the backend.

/// </summary>

Socket _socket = default!;

/// <summary>

/// The physical connection stream to the backend, without anything on top.

/// </summary>

NetworkStream _baseStream = default!;

/// <summary>

/// The physical connection stream to the backend, layered with an SSL/TLS stream if in secure mode.

/// </summary>

Stream _stream = default!;

/// <summary>

/// The parsed connection string.

/// </summary>

public NpgsqlConnectionStringBuilder Settings { get; }

Action<SslClientAuthenticationOptions>? SslClientAuthenticationOptionsCallback { get; }

#pragma warning disable CS0618 // ProvidePasswordCallback is obsolete

ProvidePasswordCallback? ProvidePasswordCallback { get; }

#pragma warning restore CS0618

Action<NegotiateAuthenticationClientOptions>? NegotiateOptionsCallback { get; }

public Encoding TextEncoding { get; private set; } = default!;

/// <summary>

/// Same as <see cref="TextEncoding"/>, except that it does not throw an exception if an invalid char is

/// encountered (exception fallback), but rather replaces it with a question mark character (replacement

/// fallback).

/// </summary>

internal Encoding RelaxedTextEncoding { get; private set; } = default!;

/// <summary>

/// Buffer used for reading data.

/// </summary>

internal NpgsqlReadBuffer ReadBuffer { get; private set; } = default!;

/// <summary>

/// If we read a data row that's bigger than <see cref="ReadBuffer"/>, we allocate an oversize buffer.

/// The original (smaller) buffer is stored here, and restored when the connection is reset.

/// </summary>

NpgsqlReadBuffer? _origReadBuffer;

/// <summary>

/// Buffer used for writing data.

/// </summary>

internal NpgsqlWriteBuffer WriteBuffer { get; private set; } = default!;

/// <summary>

/// The secret key of the backend for this connector, used for query cancellation.

/// </summary>

int _backendSecretKey;

/// <summary>

/// The process ID of the backend for this connector.

/// </summary>

internal int BackendProcessId { get; private set; }

string? _inferredUserName;

/// <summary>

/// The user name that has been inferred when the connector was opened

/// </summary>

internal string InferredUserName

{

get => _inferredUserName ?? throw new InvalidOperationException($"{nameof(InferredUserName)} cannot be accessed before the connector has been opened.");

private set => _inferredUserName = value;

}

bool SupportsPostgresCancellation => BackendProcessId != 0;

/// <summary>

/// A unique ID identifying this connector, used for logging. Currently mapped to BackendProcessId

/// </summary>

internal int Id => BackendProcessId;

internal PgSerializerOptions SerializerOptions { get; set; } = default!;

/// <summary>

/// Information about PostgreSQL and PostgreSQL-like databases (e.g. type definitions, capabilities...).

/// </summary>

public NpgsqlDatabaseInfo DatabaseInfo { get; internal set; } = default!;

/// <summary>

/// The current transaction status for this connector.

/// </summary>

internal TransactionStatus TransactionStatus { get; set; }

/// <summary>

/// A transaction object for this connector. Since only one transaction can be in progress at any given time,

/// this instance is recycled. To check whether a transaction is currently in progress on this connector,

/// see <see cref="TransactionStatus"/>.

/// </summary>

internal NpgsqlTransaction? Transaction { get; set; }

internal NpgsqlTransaction? UnboundTransaction { get; set; }

/// <summary>

/// The NpgsqlConnection that (currently) owns this connector. Null if the connector isn't

/// owned (i.e. idle in the pool)

/// </summary>

internal NpgsqlConnection? Connection { get; set; }

/// <summary>

/// The number of messages that were prepended to the current message chain, but not yet sent.

/// Note that this only tracks messages which produce a ReadyForQuery message

/// </summary>

internal int PendingPrependedResponses { get; set; }

/// <summary>

/// A ManualResetEventSlim used to make sure a cancellation request doesn't run

/// while we're reading responses for the prepended query

/// as we can't gracefully handle their cancellation.

/// </summary>

readonly ManualResetEventSlim ReadingPrependedMessagesMRE = new(initialState: true);

internal NpgsqlDataReader? CurrentReader;

internal PreparedStatementManager PreparedStatementManager { get; }

internal SqlQueryParser SqlQueryParser { get; } = new();

/// <summary>

/// If the connector is currently in COPY mode, holds a reference to the importer/exporter object.

/// Otherwise null.

/// </summary>

internal ICancelable? CurrentCopyOperation;

/// <summary>

/// Holds all run-time parameters received from the backend (via ParameterStatus messages)

/// </summary>

internal Dictionary<string, string> PostgresParameters { get; }

/// <summary>

/// Holds all run-time parameters in raw, binary format for efficient handling without allocations.

/// </summary>

readonly List<(byte[] Name, byte[] Value)> _rawParameters = [];

/// <summary>

/// If this connector was broken, this contains the exception that caused the break.

/// </summary>

volatile Exception? _breakReason;

/// <summary>

/// <para>

/// Used by the pool to indicate that I/O is currently in progress on this connector, so that another write

/// isn't started concurrently. Note that since we have only one write loop, this is only ever usedto

/// protect against an over-capacity writes into a connector that's currently *asynchronously* writing.

/// </para>

/// <para>

/// It is guaranteed that the currently-executing

/// Specifically, reading may occur - and the connector may even be returned to the pool - before this is

/// released.

/// </para>

/// </summary>

internal volatile int MultiplexAsyncWritingLock;

/// <seealso cref="MultiplexAsyncWritingLock"/>

internal void FlagAsNotWritableForMultiplexing()

{

Debug.Assert(Settings.Multiplexing);

Debug.Assert(CommandsInFlightCount > 0 || IsBroken || IsClosed,

$"About to mark multiplexing connector as non-writable, but {nameof(CommandsInFlightCount)} is {CommandsInFlightCount}");

Interlocked.Exchange(ref MultiplexAsyncWritingLock, 1);

}

/// <seealso cref="MultiplexAsyncWritingLock"/>

internal void FlagAsWritableForMultiplexing()

{

Debug.Assert(Settings.Multiplexing);

if (Interlocked.CompareExchange(ref MultiplexAsyncWritingLock, 0, 1) != 1)

throw new Exception("Multiplexing lock was not taken when releasing. Please report a bug.");

}

/// <summary>

/// A lock that's taken while a cancellation is being delivered; new queries are blocked until the

/// cancellation is delivered. This reduces the chance that a cancellation meant for a previous

/// command will accidentally cancel a later one, see #615.

/// </summary>

object CancelLock { get; } = new();

/// <summary>

/// A lock that's taken to make sure no other concurrent operation is running.

/// Break takes it to set the state of the connector.

/// Anyone else should immediately check the state and exit

/// if the connector is closed.

/// </summary>

object SyncObj { get; } = new();

/// <summary>

/// A lock that's used to wait for the Cleanup to complete while breaking the connection.

/// </summary>

object CleanupLock { get; } = new();

readonly bool _isKeepAliveEnabled;

readonly Timer? _keepAliveTimer;

/// <summary>

/// The command currently being executed by the connector, null otherwise.

/// Used only for concurrent use error reporting purposes.

/// </summary>

NpgsqlCommand? _currentCommand;

bool _sendResetOnClose;

/// <summary>

/// The connector source (e.g. pool) from where this connector came, and to which it will be returned.

/// Note that in multi-host scenarios, this references the host-specific <see cref="PoolingDataSource"/> rather than the

/// <see cref="NpgsqlMultiHostDataSource"/>.

/// </summary>

internal NpgsqlDataSource DataSource { get; }

internal string UserFacingConnectionString => DataSource.ConnectionString;

/// <summary>

/// Contains the UTC timestamp when this connector was opened, used to implement

/// <see cref="NpgsqlConnectionStringBuilder.ConnectionLifetime"/>.

/// </summary>

internal DateTime OpenTimestamp { get; private set; }

internal int ClearCounter { get; set; }

volatile bool _postgresCancellationPerformed;

internal bool PostgresCancellationPerformed

{

get => _postgresCancellationPerformed;

private set => _postgresCancellationPerformed = value;

}

volatile bool _userCancellationRequested;

CancellationTokenRegistration _cancellationTokenRegistration;

internal bool UserCancellationRequested => _userCancellationRequested;

internal CancellationToken UserCancellationToken { get; set; }

internal bool AttemptPostgresCancellation { get; private set; }

static readonly TimeSpan _cancelImmediatelyTimeout = TimeSpan.FromMilliseconds(-1);

static readonly SslApplicationProtocol _alpnProtocol = new("postgresql");

#pragma warning disable CA1859

// We're casting to IDisposable to not explicitly reference X509Certificate2 for NativeAOT

// TODO: probably pointless now, needs to be rechecked

IDisposable? _certificate;

#pragma warning restore CA1859

internal NpgsqlLoggingConfiguration LoggingConfiguration { get; }

internal ILogger ConnectionLogger { get; }

internal ILogger CommandLogger { get; }

internal ILogger TransactionLogger { get; }

internal ILogger CopyLogger { get; }

internal readonly Stopwatch QueryLogStopWatch = new();

internal EndPoint? ConnectedEndPoint { get; private set; }

#endregion

#region Constants

/// <summary>

/// The minimum timeout that can be set on internal commands such as COMMIT, ROLLBACK.

/// </summary>

/// <remarks>Precision is seconds</remarks>

internal const int MinimumInternalCommandTimeout = 3;

#endregion

#region Reusable Message Objects

byte[]? _resetWithoutDeallocateMessage;

int _resetWithoutDeallocateResponseCount;

// Backend

readonly CommandCompleteMessage _commandCompleteMessage = new();

readonly ReadyForQueryMessage _readyForQueryMessage = new();

readonly ParameterDescriptionMessage _parameterDescriptionMessage = new();

readonly DataRowMessage _dataRowMessage = new();

readonly RowDescriptionMessage _rowDescriptionMessage = new(connectorOwned: true);

// Since COPY is rarely used, allocate these lazily

CopyInResponseMessage? _copyInResponseMessage;

CopyOutResponseMessage? _copyOutResponseMessage;

CopyDataMessage? _copyDataMessage;

CopyBothResponseMessage? _copyBothResponseMessage;

#endregion

internal NpgsqlDataReader DataReader { get; set; }

internal NpgsqlDataReader? UnboundDataReader { get; set; }

#region Constructors

internal NpgsqlConnector(NpgsqlDataSource dataSource, NpgsqlConnection conn)

: this(dataSource)

{

var sslClientAuthenticationOptionsCallback = conn.SslClientAuthenticationOptionsCallback;

#pragma warning disable CS0618 // Obsolete

var provideClientCertificatesCallback = conn.ProvideClientCertificatesCallback;

var userCertificateValidationCallback = conn.UserCertificateValidationCallback;

if (provideClientCertificatesCallback is not null ||

userCertificateValidationCallback is not null)

{

if (sslClientAuthenticationOptionsCallback is not null)

throw new NotSupportedException(NpgsqlStrings.SslClientAuthenticationOptionsCallbackWithOtherCallbacksNotSupported);

sslClientAuthenticationOptionsCallback = options =>

{

if (provideClientCertificatesCallback is not null)

{

options.ClientCertificates ??= new X509Certificate2Collection();

provideClientCertificatesCallback.Invoke(options.ClientCertificates);

}

if (userCertificateValidationCallback is not null)

{

options.RemoteCertificateValidationCallback = userCertificateValidationCallback;

}

};

}

if (sslClientAuthenticationOptionsCallback is not null)

SslClientAuthenticationOptionsCallback = sslClientAuthenticationOptionsCallback;

ProvidePasswordCallback = conn.ProvidePasswordCallback;

#pragma warning restore CS0618

}

NpgsqlConnector(NpgsqlConnector connector)

: this(connector.DataSource)

{

SslClientAuthenticationOptionsCallback = connector.SslClientAuthenticationOptionsCallback;

ProvidePasswordCallback = connector.ProvidePasswordCallback;

}

NpgsqlConnector(NpgsqlDataSource dataSource)

{

Debug.Assert(dataSource.OwnsConnectors);

DataSource = dataSource;

LoggingConfiguration = dataSource.LoggingConfiguration;

ConnectionLogger = LoggingConfiguration.ConnectionLogger;

CommandLogger = LoggingConfiguration.CommandLogger;

TransactionLogger = LoggingConfiguration.TransactionLogger;

CopyLogger = LoggingConfiguration.CopyLogger;

SslClientAuthenticationOptionsCallback = dataSource.SslClientAuthenticationOptionsCallback;

NegotiateOptionsCallback = dataSource.Configuration.NegotiateOptionsCallback;

State = ConnectorState.Closed;

TransactionStatus = TransactionStatus.Idle;

Settings = dataSource.Settings;

PostgresParameters = new Dictionary<string, string>();

_isKeepAliveEnabled = Settings.KeepAlive > 0;

if (_isKeepAliveEnabled)

_keepAliveTimer = new Timer(PerformKeepAlive, null, Timeout.Infinite, Timeout.Infinite);

DataReader = new NpgsqlDataReader(this);

// TODO: Not just for automatic preparation anymore...

PreparedStatementManager = new PreparedStatementManager(this);

if (Settings.Multiplexing)

{

// Note: It's OK for this channel to be unbounded: each command enqueued to it is accompanied by sending

// it to PostgreSQL. If we overload it, a TCP zero window will make us block on the networking side

// anyway.

// Note: the in-flight channel can probably be single-writer, but that doesn't actually do anything

// at this point. And we currently rely on being able to complete the channel at any point (from

// Break). We may want to revisit this if an optimized, SingleWriter implementation is introduced.

var commandsInFlightChannel = Channel.CreateUnbounded<NpgsqlCommand>(

new UnboundedChannelOptions { SingleReader = true });

CommandsInFlightReader = commandsInFlightChannel.Reader;

CommandsInFlightWriter = commandsInFlightChannel.Writer;

// TODO: Properly implement this

if (_isKeepAliveEnabled)

throw new NotImplementedException("Keepalive not yet implemented for multiplexing");

}

}

#endregion

#region Configuration settings

internal string Host => Settings.Host!;

internal int Port => Settings.Port;

internal string Database => Settings.Database!;

string KerberosServiceName => Settings.KerberosServiceName;

int ConnectionTimeout => Settings.Timeout;

#endregion Configuration settings

#region State management

int _state;

/// <summary>

/// Gets the current state of the connector

/// </summary>

internal ConnectorState State

{

get => (ConnectorState)_state;

set

{

var newState = (int)value;

if (newState == _state)

return;

if (newState is < 0 or > (int)ConnectorState.Replication)

ThrowHelper.ThrowArgumentOutOfRangeException(nameof(value), "Unknown state: " + value);

Interlocked.Exchange(ref _state, newState);

}

}

/// <summary>

/// Returns whether the connector is open, regardless of any task it is currently performing

/// </summary>

bool IsConnected => State is not (ConnectorState.Closed or ConnectorState.Connecting or ConnectorState.Broken);

internal bool IsReady => State == ConnectorState.Ready;

internal bool IsClosed => State == ConnectorState.Closed;

internal bool IsBroken => State == ConnectorState.Broken;

#endregion

#region Open

/// <summary>

/// Opens the physical connection to the server.

/// </summary>

/// <remarks>Usually called by the RequestConnector

/// Method of the connection pool manager.</remarks>

internal async Task Open(NpgsqlTimeout timeout, bool async, CancellationToken cancellationToken)

{

Debug.Assert(State == ConnectorState.Closed);

State = ConnectorState.Connecting;

LogMessages.OpeningPhysicalConnection(ConnectionLogger, Host, Port, Database, UserFacingConnectionString);

var startOpenTimestamp = Stopwatch.GetTimestamp();

Activity? activity = null;

try

{

var username = await GetUsernameAsync(async, cancellationToken).ConfigureAwait(false);

activity = NpgsqlActivitySource.ConnectionOpen(this);

var gssEncMode = GetGssEncMode(Settings);

await OpenCore(this, username, Settings.SslMode, gssEncMode, timeout, async, cancellationToken).ConfigureAwait(false);

if (activity is not null)

NpgsqlActivitySource.Enrich(activity, this);

await DataSource.Bootstrap(this, timeout, forceReload: false, async, cancellationToken).ConfigureAwait(false);

Debug.Assert(DataSource.SerializerOptions is not null);

Debug.Assert(DataSource.DatabaseInfo is not null);

SerializerOptions = DataSource.SerializerOptions;

DatabaseInfo = DataSource.DatabaseInfo;

if (Settings.Pooling && Settings is { Multiplexing: false, NoResetOnClose: false } && DatabaseInfo.SupportsDiscard)

{

_sendResetOnClose = true;

GenerateResetMessage();

}

OpenTimestamp = DateTime.UtcNow;

if (Settings.Multiplexing)

{

// Start an infinite async loop, which processes incoming multiplexing traffic.

// It is intentionally not awaited and will run as long as the connector is alive.

// The CommandsInFlightWriter channel is completed in Cleanup, which should cause this task

// to complete.

// Make sure we do not flow AsyncLocals like Activity.Current

using var __ = ExecutionContext.SuppressFlow();

_ = Task.Run(MultiplexingReadLoop, CancellationToken.None)

.ContinueWith(t =>

{

// Note that we *must* observe the exception if the task is faulted.

ConnectionLogger.LogError(t.Exception!, "Exception bubbled out of multiplexing read loop", Id);

}, TaskContinuationOptions.OnlyOnFaulted);

}

if (_isKeepAliveEnabled)

{

// Start the keep alive mechanism to work by scheduling the timer.

// Otherwise, it doesn't work for cases when no query executed during

// the connection lifetime in case of a new connector.

lock (SyncObj)

{

var keepAlive = Settings.KeepAlive * 1000;

_keepAliveTimer!.Change(keepAlive, keepAlive);

}

}

if (DataSource.ConnectionInitializerAsync is not null)

{

Debug.Assert(DataSource.ConnectionInitializer is not null);

var tempConnection = new NpgsqlConnection(DataSource, this);

try

{

if (async)

await DataSource.ConnectionInitializerAsync(tempConnection).ConfigureAwait(false);

else

DataSource.ConnectionInitializer(tempConnection);

}

finally

{

// Note that we can't just close/dispose the NpgsqlConnection, since that puts the connector back in the pool.

// But we transition it to disposed immediately, in case the user decides to capture the NpgsqlConnection and use it

// later.

Connection?.MakeDisposed();

Connection = null;

}

}

if (activity is not null)

NpgsqlActivitySource.CommandStop(activity);

LogMessages.OpenedPhysicalConnection(

ConnectionLogger, Host, Port, Database, UserFacingConnectionString,

(long)Stopwatch.GetElapsedTime(startOpenTimestamp).TotalMilliseconds, Id);

}

catch (Exception e)

{

if (activity is not null)

NpgsqlActivitySource.SetException(activity, e);

Break(e);

throw;

}

static async Task OpenCore(

NpgsqlConnector conn,

string username,

SslMode sslMode,

GssEncryptionMode gssEncMode,

NpgsqlTimeout timeout,

bool async,

CancellationToken cancellationToken)

{

await conn.RawOpen(sslMode, gssEncMode, timeout, async, cancellationToken).ConfigureAwait(false);

timeout.CheckAndApply(conn);

conn.WriteStartupMessage(username);

await conn.Flush(async, cancellationToken).ConfigureAwait(false);

using var cancellationRegistration = conn.StartCancellableOperation(cancellationToken, attemptPgCancellation: false);

try

{

await conn.Authenticate(username, timeout, async, cancellationToken).ConfigureAwait(false);

}

catch (Exception e) when

// Any error after trying with GSS encryption

(gssEncMode == GssEncryptionMode.Prefer ||

// Auth error with/without SSL

(e is PostgresException { SqlState: PostgresErrorCodes.InvalidAuthorizationSpecification } &&

(sslMode == SslMode.Prefer && conn.IsSslEncrypted || sslMode == SslMode.Allow && !conn.IsSslEncrypted)))

{

if (gssEncMode == GssEncryptionMode.Prefer)

{

conn.ConnectionLogger.LogTrace(e, "Error while opening physical connection with GSS encryption, retrying without it");

gssEncMode = GssEncryptionMode.Disable;

}

else

sslMode = sslMode == SslMode.Prefer ? SslMode.Disable : SslMode.Require;

cancellationRegistration.Dispose();

Debug.Assert(!conn.IsBroken);

conn.Cleanup();

// If Prefer was specified and we failed (with SSL), retry without SSL.

// If Allow was specified and we failed (without SSL), retry with SSL

await OpenCore(

conn,

username,

sslMode,

gssEncMode,

timeout,

async,

cancellationToken).ConfigureAwait(false);

return;

}

// We treat BackendKeyData as optional because some PostgreSQL-like database

// don't send it (CockroachDB, CrateDB)

var msg = await conn.ReadMessage(async).ConfigureAwait(false);

if (msg.Code == BackendMessageCode.BackendKeyData)

{

var keyDataMsg = (BackendKeyDataMessage)msg;

conn.BackendProcessId = keyDataMsg.BackendProcessId;

conn._backendSecretKey = keyDataMsg.BackendSecretKey;

msg = await conn.ReadMessage(async).ConfigureAwait(false);

}

if (msg.Code != BackendMessageCode.ReadyForQuery)

throw new NpgsqlException($"Received backend message {msg.Code} while expecting ReadyForQuery. Please file a bug.");

conn.State = ConnectorState.Ready;

}

}

internal async ValueTask<GssEncryptionResult> GSSEncrypt(bool async, bool isRequired, CancellationToken cancellationToken)

{

ConnectionLogger.LogTrace("Negotiating GSS encryption");

var targetName = $"{KerberosServiceName}/{Host}";

var clientOptions = new NegotiateAuthenticationClientOptions { TargetName = targetName };

NegotiateOptionsCallback?.Invoke(clientOptions);

var authentication = new NegotiateAuthentication(clientOptions);

try

{

var data = authentication.GetOutgoingBlob(ReadOnlySpan<byte>.Empty, out var statusCode)!;

if (statusCode != NegotiateAuthenticationStatusCode.ContinueNeeded)

{

// Unable to retrieve credentials

// If it's required, throw an appropriate exception

if (isRequired)

throw new NpgsqlException($"Unable to negotiate GSS encryption: {statusCode}");

return GssEncryptionResult.GetCredentialFailure;

}

WriteGSSEncryptRequest();

await Flush(async, cancellationToken).ConfigureAwait(false);

await ReadBuffer.Ensure(1, async).ConfigureAwait(false);

var response = (char)ReadBuffer.ReadByte();

// TODO: Server can respond with an error here

// but according to documentation we shouldn't display this error to the user/application

// since the server has not been authenticated (CVE-2024-10977)

// See https://www.postgresql.org/docs/current/protocol-flow.html#PROTOCOL-FLOW-GSSAPI

switch (response)

{

default:

throw new NpgsqlException($"Received unknown response {response} for GSSEncRequest (expecting G or N)");

case 'N':

return GssEncryptionResult.NegotiateFailure;

case 'G':

break;

}

if (ReadBuffer.ReadBytesLeft > 0)

throw new NpgsqlException(

"Additional unencrypted data received after GSS encryption negotiation - this should never happen, and may be an indication of a man-in-the-middle attack.");

var lengthBuffer = new byte[4];

await WriteGssEncryptMessage(async, data, lengthBuffer).ConfigureAwait(false);

while (true)

{

if (async)

await _stream.ReadExactlyAsync(lengthBuffer, cancellationToken).ConfigureAwait(false);

else

_stream.ReadExactly(lengthBuffer);

var messageLength = BitConverter.IsLittleEndian

? BinaryPrimitives.ReverseEndianness(Unsafe.ReadUnaligned<int>(ref lengthBuffer[0]))

: Unsafe.ReadUnaligned<int>(ref lengthBuffer[0]);

var buffer = ArrayPool<byte>.Shared.Rent(messageLength);

if (async)

await _stream.ReadExactlyAsync(buffer.AsMemory(0, messageLength), cancellationToken).ConfigureAwait(false);

else

_stream.ReadExactly(buffer.AsSpan(0, messageLength));

data = authentication.GetOutgoingBlob(buffer.AsSpan(0, messageLength), out statusCode);

ArrayPool<byte>.Shared.Return(buffer, clearArray: true);

if (statusCode is not NegotiateAuthenticationStatusCode.Completed and not NegotiateAuthenticationStatusCode.ContinueNeeded)

throw new NpgsqlException($"Error while negotiating GSS encryption: {statusCode}");

// TODO: the code below is the copy from GSS/SSPI auth

// It's unknown whether it holds true here or not

// We might get NegotiateAuthenticationStatusCode.Completed but the data will not be null

// This can happen if it's the first cycle, in which case we have to send that data to complete handshake (#4888)

if (data is null)

{

Debug.Assert(statusCode == NegotiateAuthenticationStatusCode.Completed);

break;

}

await WriteGssEncryptMessage(async, data, lengthBuffer).ConfigureAwait(false);

}

_stream = new GSSStream(_stream, authentication);

ReadBuffer.Underlying = _stream;

WriteBuffer.Underlying = _stream;

IsGssEncrypted = true;

authentication = null;

ConnectionLogger.LogTrace("GSS encryption successful");

return GssEncryptionResult.Success;

async ValueTask WriteGssEncryptMessage(bool async, byte[] data, byte[] lengthBuffer)

{

BinaryPrimitives.WriteInt32BigEndian(lengthBuffer, data.Length);

if (async)

{

await _stream.WriteAsync(lengthBuffer, cancellationToken).ConfigureAwait(false);

await _stream.WriteAsync(data, cancellationToken).ConfigureAwait(false);

await _stream.FlushAsync(cancellationToken).ConfigureAwait(false);

}

else

{

_stream.Write(lengthBuffer);

_stream.Write(data);

_stream.Flush();

}

}

}

catch (Exception e)

{

throw new NpgsqlException("Exception while performing GSS encryption", e);

}

finally

{

authentication?.Dispose();

}

}

internal async ValueTask<DatabaseState> QueryDatabaseState(

NpgsqlTimeout timeout, bool async, CancellationToken cancellationToken = default)

{

using var batch = CreateBatch();

batch.BatchCommands.Add(new NpgsqlBatchCommand("select pg_is_in_recovery()"));

batch.BatchCommands.Add(new NpgsqlBatchCommand("SHOW default_transaction_read_only"));

batch.Timeout = (int)timeout.CheckAndGetTimeLeft().TotalSeconds;

var reader = async ? await batch.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false) : batch.ExecuteReader();

try

{

if (async)

{

await reader.ReadAsync(cancellationToken).ConfigureAwait(false);

_isHotStandBy = reader.GetBoolean(0);

await reader.NextResultAsync(cancellationToken).ConfigureAwait(false);

await reader.ReadAsync(cancellationToken).ConfigureAwait(false);

}

else

{

reader.Read();

_isHotStandBy = reader.GetBoolean(0);

reader.NextResult();

reader.Read();

}

_isTransactionReadOnly = reader.GetString(0) != "off";

var databaseState = UpdateDatabaseState();

Debug.Assert(databaseState.HasValue);

return databaseState.Value;

}

finally

{

if (async)

await reader.DisposeAsync().ConfigureAwait(false);

else

reader.Dispose();

}

}

void WriteStartupMessage(string username)

{

var startupParams = new Dictionary<string, string>

{

["user"] = username,

["client_encoding"] = Settings.ClientEncoding ??

PostgresEnvironment.ClientEncoding ??

"UTF8"

};

if (Settings.Database is not null)

startupParams["database"] = Settings.Database;

var applicationName = Settings.ApplicationName ?? PostgresEnvironment.AppName;

if (applicationName?.Length > 0)

startupParams["application_name"] = applicationName;

if (Settings.SearchPath?.Length > 0)

startupParams["search_path"] = Settings.SearchPath;

var timezone = Settings.Timezone ?? PostgresEnvironment.TimeZone;

if (timezone != null)

startupParams["TimeZone"] = timezone;

var options = Settings.Options ?? PostgresEnvironment.Options;

if (options?.Length > 0)

startupParams["options"] = options;

switch (Settings.ReplicationMode)

{

case ReplicationMode.Logical:

startupParams["replication"] = "database";

break;

case ReplicationMode.Physical:

startupParams["replication"] = "true";

break;

}

WriteStartup(startupParams);

}

ValueTask<string> GetUsernameAsync(bool async, CancellationToken cancellationToken)

{

var username = Settings.Username;

if (username?.Length > 0)

{

InferredUserName = username;

return new(username);

}

username = PostgresEnvironment.User;

if (username?.Length > 0)

{

InferredUserName = username;

return new(username);

}

return GetUsernameAsyncInternal();

async ValueTask<string> GetUsernameAsyncInternal()

{

if (!RuntimeInformation.IsOSPlatform(OSPlatform.Windows))

{

username = await DataSource.IntegratedSecurityHandler.GetUsername(async, Settings.IncludeRealm, ConnectionLogger,

cancellationToken).ConfigureAwait(false);

if (username?.Length > 0)

{

InferredUserName = username;

return username;

}

}

username = Environment.UserName;

if (username?.Length > 0)

{

InferredUserName = username;

return username;

}

throw new NpgsqlException("No username could be found, please specify one explicitly");

}

}

async Task RawOpen(SslMode sslMode, GssEncryptionMode gssEncryptionMode, NpgsqlTimeout timeout, bool async, CancellationToken cancellationToken)

{

try

{

if (async)

await ConnectAsync(timeout, cancellationToken).ConfigureAwait(false);

else

Connect(timeout);

ConnectionLogger.LogTrace("Socket connected to {Host}:{Port}", Host, Port);

_baseStream = new NetworkStream(_socket, true);

_stream = _baseStream;

if (Settings.Encoding == "UTF8")

{

TextEncoding = NpgsqlWriteBuffer.UTF8Encoding;

RelaxedTextEncoding = NpgsqlWriteBuffer.RelaxedUTF8Encoding;

}

else

{

TextEncoding = Encoding.GetEncoding(Settings.Encoding, EncoderFallback.ExceptionFallback, DecoderFallback.ExceptionFallback);

RelaxedTextEncoding = Encoding.GetEncoding(Settings.Encoding, EncoderFallback.ReplacementFallback, DecoderFallback.ReplacementFallback);

}

ReadBuffer = new NpgsqlReadBuffer(this, _stream, _socket, Settings.ReadBufferSize, TextEncoding, RelaxedTextEncoding);

WriteBuffer = new NpgsqlWriteBuffer(this, _stream, _socket, Settings.WriteBufferSize, TextEncoding);

timeout.CheckAndApply(this);

IsSslEncrypted = false;

IsGssEncrypted = false;

var gssEncryptResult = await TryNegotiateGssEncryption(gssEncryptionMode, async, cancellationToken).ConfigureAwait(false);

if (gssEncryptResult == GssEncryptionResult.Success)

return;

timeout.CheckAndApply(this);

if (GetSslNegotiation(Settings) == SslNegotiation.Direct)

{

// We already check that in NpgsqlConnectionStringBuilder.PostProcessAndValidate, but since we also allow environment variables...

if (Settings.SslMode is not SslMode.Require and not SslMode.VerifyCA and not SslMode.VerifyFull)

throw new ArgumentException("SSL Mode has to be Require or higher to be used with direct SSL Negotiation");

if (gssEncryptResult == GssEncryptionResult.NegotiateFailure)

{

// We can be here only if it's fallback from preferred (but failed) gss encryption

// In this case, direct encryption isn't going to work anymore, so we throw a bogus exception to retry again without gss

// Alternatively, we can instead just go with the usual route of writing SslRequest, ignoring direct ssl

// But this is how libpq works

Debug.Assert(gssEncryptionMode == GssEncryptionMode.Prefer);

// The exception message doesn't matter since we're going to retry again

throw new NpgsqlException();

}

await DataSource.TransportSecurityHandler.NegotiateEncryption(async, this, sslMode, timeout, cancellationToken).ConfigureAwait(false);

if (ReadBuffer.ReadBytesLeft > 0)

throw new NpgsqlException("Additional unencrypted data received after SSL negotiation - this should never happen, and may be an indication of a man-in-the-middle attack.");

}

else if ((sslMode is SslMode.Prefer && DataSource.TransportSecurityHandler.SupportEncryption) ||

sslMode is SslMode.Require or SslMode.VerifyCA or SslMode.VerifyFull)

{

WriteSslRequest();

await Flush(async, cancellationToken).ConfigureAwait(false);

await ReadBuffer.Ensure(1, async).ConfigureAwait(false);

var response = (char)ReadBuffer.ReadByte();

timeout.CheckAndApply(this);

switch (response)

{

default:

throw new NpgsqlException($"Received unknown response {response} for SSLRequest (expecting S or N)");

case 'N':

if (sslMode != SslMode.Prefer)

throw new NpgsqlException("SSL connection requested. No SSL enabled connection from this host is configured.");

break;

case 'S':

await DataSource.TransportSecurityHandler.NegotiateEncryption(async, this, sslMode, timeout, cancellationToken).ConfigureAwait(false);

break;

}

if (ReadBuffer.ReadBytesLeft > 0)

throw new NpgsqlException("Additional unencrypted data received after SSL negotiation - this should never happen, and may be an indication of a man-in-the-middle attack.");

}

}

catch

{

_stream?.Dispose();

_stream = null!;

_baseStream?.Dispose();

_baseStream = null!;