using System;

using System.Buffers;

using System.Collections.Generic;

using System.Data;

using System.Diagnostics;

using System.IO;

using System.Linq;

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.Logging;

using Npgsql.TypeMapping;

using Npgsql.Util;

using static Npgsql.Util.Statics;

using System.Transactions;

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>

public sealed partial class NpgsqlConnector : IDisposable

{

#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; }

ProvideClientCertificatesCallback? ProvideClientCertificatesCallback { get; }

RemoteCertificateValidationCallback? UserCertificateValidationCallback { get; }

ProvidePasswordCallback? ProvidePasswordCallback { get; }

PhysicalOpenCallback? PhysicalOpenCallback { get; set; }

PhysicalOpenAsyncCallback? PhysicalOpenAsyncCallback { get; set; }

internal 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; }

bool SupportsPostgresCancellation => BackendProcessId != 0;

/// <summary>

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

/// </summary>

internal int Id => BackendProcessId;

/// <summary>

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

/// </summary>

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

internal ConnectorTypeMapper TypeMapper { get; 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; }

internal NpgsqlDataReader? CurrentReader;

internal PreparedStatementManager PreparedStatementManager;

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 readonly Dictionary<string, string> PostgresParameters;

/// <summary>

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

/// </summary>

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

/// <summary>

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

/// </summary>

volatile Exception? _breakReason;

/// <summary>

/// Semaphore, used to synchronize DatabaseInfo between multiple connections, so it wouldn't be loaded in parallel.

/// </summary>

static readonly SemaphoreSlim DatabaseInfoSemaphore = new(1);

/// <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()

{

if (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()

{

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

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

}

/// <summary>

/// The timeout for reading messages that are part of the user's command

/// (i.e. which aren't internal prepended commands).

/// </summary>

/// <remarks>Precision is milliseconds</remarks>

internal int UserTimeout { private get; set; }

/// <summary>

/// A lock that's taken while a user action is in progress, e.g. a command being executed.

/// Only used when keepalive is enabled, otherwise null.

/// </summary>

SemaphoreSlim? _userLock;

/// <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>

internal object CancelLock { get; }

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="ConnectorPool"/> rather than the

/// <see cref="MultiHostConnectorPool"/>,

/// </summary>

readonly ConnectorSource _connectorSource;

/// <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 NpgsqlLogger Log = NpgsqlLogManager.CreateLogger(nameof(NpgsqlConnector));

internal readonly Stopwatch QueryLogStopWatch = new();

#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();

// 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(ConnectorSource connectorSource, NpgsqlConnection conn)

: this(connectorSource)

{

ProvideClientCertificatesCallback = conn.ProvideClientCertificatesCallback;

UserCertificateValidationCallback = conn.UserCertificateValidationCallback;

ProvidePasswordCallback = conn.ProvidePasswordCallback;

PhysicalOpenCallback = conn.PhysicalOpenCallback;

PhysicalOpenAsyncCallback = conn.PhysicalOpenAsyncCallback;

}

NpgsqlConnector(NpgsqlConnector connector)

: this(connector._connectorSource)

{

ProvideClientCertificatesCallback = connector.ProvideClientCertificatesCallback;

UserCertificateValidationCallback = connector.UserCertificateValidationCallback;

ProvidePasswordCallback = connector.ProvidePasswordCallback;

}

NpgsqlConnector(ConnectorSource connectorSource)

{

_connectorSource = connectorSource;

State = ConnectorState.Closed;

TransactionStatus = TransactionStatus.Idle;

Settings = connectorSource.Settings;

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

CancelLock = new object();

_isKeepAliveEnabled = Settings.KeepAlive > 0;

if (_isKeepAliveEnabled)

{

_userLock = new SemaphoreSlim(1, 1);

_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;

SslMode SslMode => Settings.SslMode;

int ConnectionTimeout => Settings.Timeout;

bool IntegratedSecurity => Settings.IntegratedSecurity;

internal bool ConvertInfinityDateTime => Settings.ConvertInfinityDateTime;

/// <summary>

/// The actual command timeout value that gets set on internal commands.

/// </summary>

/// <remarks>Precision is milliseconds</remarks>

int InternalCommandTimeout

{

get

{

var internalTimeout = Settings.InternalCommandTimeout;

if (internalTimeout == -1)

return Math.Max(Settings.CommandTimeout, MinimumInternalCommandTimeout) * 1000;

// Todo: Decide what we really want here

// This assertion can easily fail if InternalCommandTimeout is set to 1 or 2 in the connection string

// We probably don't want to allow these values but in that case a Debug.Assert is the wrong way to enforce it.

Debug.Assert(internalTimeout == 0 || internalTimeout >= MinimumInternalCommandTimeout);

return internalTimeout * 1000;

}

}

#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;

Interlocked.Exchange(ref _state, newState);

}

}

/// <summary>

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

/// </summary>

bool IsConnected

=> State switch

{

ConnectorState.Ready => true,

ConnectorState.Executing => true,

ConnectorState.Fetching => true,

ConnectorState.Waiting => true,

ConnectorState.Copy => true,

ConnectorState.Replication => true,

ConnectorState.Closed => false,

ConnectorState.Connecting => false,

ConnectorState.Broken => false,

_ => throw new ArgumentOutOfRangeException("Unknown state: " + State)

};

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;

try

{

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

var username = GetUsername();

if (Settings.Database == null)

Settings.Database = username;

timeout.CheckAndApply(this);

WriteStartupMessage(username);

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

using (StartCancellableOperation(cancellationToken, attemptPgCancellation: false))

{

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

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

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

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

if (msg.Code == BackendMessageCode.BackendKeyData)

{

var keyDataMsg = (BackendKeyDataMessage)msg;

BackendProcessId = keyDataMsg.BackendProcessId;

_backendSecretKey = keyDataMsg.BackendSecretKey;

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

}

if (msg.Code != BackendMessageCode.ReadyForQuery)

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

State = ConnectorState.Ready;

}

await LoadDatabaseInfo(forceReload: false, timeout, async, cancellationToken).ConfigureAwait(false);

if (Settings.Pooling && !Settings.Multiplexing && !Settings.NoResetOnClose && DatabaseInfo.SupportsDiscard)

{

_sendResetOnClose = true;

GenerateResetMessage();

}

OpenTimestamp = DateTime.UtcNow;

Log.Trace($"Opened connection to {Host}:{Port}");

if (async && PhysicalOpenAsyncCallback is not null)

await PhysicalOpenAsyncCallback(this).ConfigureAwait(false);

else if (!async && PhysicalOpenCallback is not null)

PhysicalOpenCallback(this);

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.

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

.ContinueWith(t =>

{

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

Log.Error("Exception bubbled out of multiplexing read loop", t.Exception!, 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 (this)

{

var keepAlive = Settings.KeepAlive * 1000;

_keepAliveTimer!.Change(keepAlive, keepAlive);

}

}

}

catch (Exception e)

{

ClusterStateCache.UpdateClusterState(Settings.Host!, Settings.Port, ClusterState.Offline,

DateTime.UtcNow, Settings.HostRecheckSecondsTranslated);

Break(e);

throw;

}

}

internal async ValueTask LoadDatabaseInfo(bool forceReload, NpgsqlTimeout timeout, bool async,

CancellationToken cancellationToken = default)

{

// The type loading below will need to send queries to the database, and that depends on a type mapper

// being set up (even if its empty)

TypeMapper = new ConnectorTypeMapper(this);

var key = new NpgsqlDatabaseInfoCacheKey(Settings);

if (forceReload || !NpgsqlDatabaseInfo.Cache.TryGetValue(key, out var database))

{

var hasSemaphore = async

? await DatabaseInfoSemaphore.WaitAsync(timeout.CheckAndGetTimeLeft(), cancellationToken).ConfigureAwait(false)

: DatabaseInfoSemaphore.Wait(timeout.CheckAndGetTimeLeft(), cancellationToken);

// We've timed out - calling Check, to throw the correct exception

if (!hasSemaphore)

timeout.Check();

try

{

if (forceReload || !NpgsqlDatabaseInfo.Cache.TryGetValue(key, out database))

{

NpgsqlDatabaseInfo.Cache[key] = database = await NpgsqlDatabaseInfo.Load(this, timeout, async).ConfigureAwait(false);

}

}

finally

{

DatabaseInfoSemaphore.Release();

}

}

DatabaseInfo = TypeMapper.DatabaseInfo = database!;

TypeMapper.Reset();

}

internal async ValueTask<ClusterState> QueryClusterState(

NpgsqlTimeout timeout, bool async, CancellationToken cancellationToken = default)

{

using var cmd = CreateCommand("select pg_is_in_recovery(); SHOW default_transaction_read_only");

cmd.CommandTimeout = (int)timeout.CheckAndGetTimeLeft().TotalSeconds;

// We're taking the timestamp before the query is sent, because due to issues (IO, operation ordering, etc) we can receive an

// 'old' state. Otherwise, execution of the query shouldn't make notable difference.

var timeStamp = DateTime.UtcNow;

var reader = async ? await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false) : cmd.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 clusterState = UpdateClusterState(ignoreDatabaseVersion: false);

Debug.Assert(clusterState.HasValue);

return clusterState.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",

["database"] = Settings.Database!

};

if (Settings.ApplicationName?.Length > 0)

startupParams["application_name"] = Settings.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);

}

string GetUsername()

{

var username = Settings.Username;

if (username?.Length > 0)

return username;

username = PostgresEnvironment.User;

if (username?.Length > 0)

return username;

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

{

username = KerberosUsernameProvider.GetUsername(Settings.IncludeRealm);

if (username?.Length > 0)

return username;

}

username = Environment.UserName;

if (username?.Length > 0)

return username;

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

}

async Task RawOpen(NpgsqlTimeout timeout, bool async, CancellationToken cancellationToken)

{

var cert = default(X509Certificate2?);

try

{

if (async)

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

else

Connect(timeout);

_baseStream = new NetworkStream(_socket, true);

_stream = _baseStream;

if (Settings.Encoding == "UTF8")

{

TextEncoding = PGUtil.UTF8Encoding;

RelaxedTextEncoding = PGUtil.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);

if (SslMode == SslMode.Require || SslMode == SslMode.Prefer)

{

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.Require)

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

break;

case 'S':

var clientCertificates = new X509Certificate2Collection();

var certPath = Settings.SslCertificate ?? PostgresEnvironment.SslCert ?? PostgresEnvironment.SslCertDefault;

if (certPath != null)

{

var password = Settings.SslPassword;

if (Path.GetExtension(certPath).ToUpperInvariant() != ".PFX")

{

#if NET5_0_OR_GREATER

// It's PEM time

var keyPath = Settings.SslKey ?? PostgresEnvironment.SslKey ?? PostgresEnvironment.SslKeyDefault;

cert = string.IsNullOrEmpty(password)

? X509Certificate2.CreateFromPemFile(certPath, keyPath)

: X509Certificate2.CreateFromEncryptedPemFile(certPath, password, keyPath);

if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))

{

// Windows crypto API has a bug with pem certs

// See #3650

using var previousCert = cert;

cert = new X509Certificate2(cert.Export(X509ContentType.Pkcs12));

}

#else

throw new NotSupportedException("PEM certificates are only supported with .NET 5 and higher");

#endif

}

if (cert is null)

cert = new X509Certificate2(certPath, password);

clientCertificates.Add(cert);

}

ProvideClientCertificatesCallback?.Invoke(clientCertificates);

var certificateValidationCallback = Settings.TrustServerCertificate

? SslTrustServerValidation

: (Settings.RootCertificate ?? PostgresEnvironment.SslCertRoot ?? PostgresEnvironment.SslCertRootDefault) is { } certRootPath

? SslRootValidation(certRootPath)

: UserCertificateValidationCallback is { } userValidation

? userValidation

: SslDefaultValidation;

timeout.CheckAndApply(this);

try

{

var sslStream = new SslStream(_stream, leaveInnerStreamOpen: false, certificateValidationCallback);

var sslProtocols = SslProtocols.None;

// On .NET Framework SslProtocols.None can be disabled, see #3718

#if NETSTANDARD2_0

sslProtocols = SslProtocols.Tls | SslProtocols.Tls11 | SslProtocols.Tls12;

#endif

if (async)

{

await sslStream.AuthenticateAsClientAsync(

Host, clientCertificates, sslProtocols, Settings.CheckCertificateRevocation)

.ConfigureAwait(false);

}

else

{

sslStream.AuthenticateAsClient(Host, clientCertificates, sslProtocols, Settings.CheckCertificateRevocation);

}

_stream = sslStream;

}

catch (Exception e)

{

throw new NpgsqlException("Exception while performing SSL handshake", e);

}

ReadBuffer.Underlying = _stream;

WriteBuffer.Underlying = _stream;

IsSecure = true;

Log.Trace("SSL negotiation successful");

break;

}

ReadBuffer.Clear(); // Reset to empty after reading single SSL char

}

Log.Trace($"Socket connected to {Host}:{Port}");

}

catch

{

cert?.Dispose();

_stream?.Dispose();

_stream = null!;

_baseStream?.Dispose();

_baseStream = null!;

_socket?.Dispose();

_socket = null!;

throw;

}

}

void Connect(NpgsqlTimeout timeout)

{

// Note that there aren't any timeout-able or cancellable DNS methods

var endpoints = NpgsqlConnectionStringBuilder.IsUnixSocket(Host, Port, out var socketPath)

? new EndPoint[] { new UnixDomainSocketEndPoint(socketPath) }

: Dns.GetHostAddresses(Host).Select(a => new IPEndPoint(a, Port)).ToArray();

timeout.Check();

// Give each endpoint an equal share of the remaining time

var perEndpointTimeout = -1; // Default to infinity

if (timeout.IsSet)

perEndpointTimeout = (int)(timeout.CheckAndGetTimeLeft().Ticks / endpoints.Length / 10);

for (var i = 0; i < endpoints.Length; i++)

{

var endpoint = endpoints[i];

Log.Trace($"Attempting to connect to {endpoint}");

var protocolType =

endpoint.AddressFamily == AddressFamily.InterNetwork ||

endpoint.AddressFamily == AddressFamily.InterNetworkV6

? ProtocolType.Tcp

: ProtocolType.IP;

var socket = new Socket(endpoint.AddressFamily, SocketType.Stream, protocolType)

{

Blocking = false

};

try

{

try

{

socket.Connect(endpoint);

}

catch (SocketException e)

{

if (e.SocketErrorCode != SocketError.WouldBlock)

throw;

}

var write = new List<Socket> { socket };

var error = new List<Socket> { socket };

Socket.Select(null, write, error, perEndpointTimeout);

var errorCode = (int) socket.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Error)!;

if (errorCode != 0)

throw new SocketException(errorCode);

if (!write.Any())

throw new TimeoutException("Timeout during connection attempt");

socket.Blocking = true;

SetSocketOptions(socket);

_socket = socket;

return;

}

catch (Exception e)

{

try { socket.Dispose(); }

catch

{

// ignored

}

Log.Trace($"Failed to connect to {endpoint}", e);

if (i == endpoints.Length - 1)

throw new NpgsqlException($"Failed to connect to {endpoint}", e);

}

}

}

async Task ConnectAsync(NpgsqlTimeout timeout, CancellationToken cancellationToken)

{

// Note that there aren't any timeout-able or cancellable DNS methods

var endpoints = NpgsqlConnectionStringBuilder.IsUnixSocket(Host, Port, out var socketPath)

? new EndPoint[] { new UnixDomainSocketEndPoint(socketPath) }

: (await GetHostAddressesAsync(timeout, cancellationToken).ConfigureAwait(false))

.Select(a => new IPEndPoint(a, Port)).ToArray();

// Give each IP an equal share of the remaining time

var perIpTimespan = default(TimeSpan);

var perIpTimeout = timeout;

if (timeout.IsSet)

{

perIpTimespan = new TimeSpan(timeout.CheckAndGetTimeLeft().Ticks / endpoints.Length);

perIpTimeout = new NpgsqlTimeout(perIpTimespan);

}

for (var i = 0; i < endpoints.Length; i++)

{

var endpoint = endpoints[i];

Log.Trace($"Attempting to connect to {endpoint}");

var protocolType =

endpoint.AddressFamily == AddressFamily.InterNetwork ||

endpoint.AddressFamily == AddressFamily.InterNetworkV6

? ProtocolType.Tcp

: ProtocolType.IP;

var socket = new Socket(endpoint.AddressFamily, SocketType.Stream, protocolType);

try

{

await OpenSocketConnectionAsync(socket, endpoint, perIpTimeout, cancellationToken).ConfigureAwait(false);

SetSocketOptions(socket);

_socket = socket;

return;

}

catch (Exception e)

{

try

{

socket.Dispose();

}

catch

{

// ignored

}

cancellationToken.ThrowIfCancellationRequested();

if (e is OperationCanceledException)

e = new TimeoutException("Timeout during connection attempt");

Log.Trace($"Failed to connect to {endpoint}", e);

if (i == endpoints.Length - 1)

throw new NpgsqlException($"Failed to connect to {endpoint}", e);

}

}

Task<IPAddress[]> GetHostAddressesAsync(NpgsqlTimeout timeout, CancellationToken cancellationToken)

{

// .NET 6.0 added cancellation support to GetHostAddressesAsync, which allows us to implement real

// cancellation and timeout. On older TFMs, we fake-cancel the operation, i.e. stop waiting

// and raise the exception, but the actual connection task is left running.

#if NET6_0_OR_GREATER

var task = TaskExtensions.ExecuteWithTimeout(

ct => Dns.GetHostAddressesAsync(Host, ct),

timeout, cancellationToken);

#else

var task = Dns.GetHostAddressesAsync(Host);

#endif

// As the cancellation support of GetHostAddressesAsync is not guaranteed on all platforms

// we apply the fake-cancel mechanism in all cases.

return task.WithCancellationAndTimeout(timeout, cancellationToken);

}

static Task OpenSocketConnectionAsync(Socket socket, EndPoint endpoint, NpgsqlTimeout perIpTimeout, CancellationToken cancellationToken)

{

// .NET 5.0 added cancellation support to ConnectAsync, which allows us to implement real

// cancellation and timeout. On older TFMs, we fake-cancel the operation, i.e. stop waiting

// and raise the exception, but the actual connection task is left running.

#if NET5_0_OR_GREATER

return TaskExtensions.ExecuteWithTimeout(

ct => socket.ConnectAsync(endpoint, ct).AsTask(),

perIpTimeout, cancellationToken);

#else

return socket.ConnectAsync(endpoint)

.WithCancellationAndTimeout(perIpTimeout, cancellationToken);

#endif

}

}

void SetSocketOptions(Socket socket)

{

if (socket.AddressFamily == AddressFamily.InterNetwork)

socket.NoDelay = true;

if (Settings.SocketReceiveBufferSize > 0)

socket.ReceiveBufferSize = Settings.SocketReceiveBufferSize;

if (Settings.SocketSendBufferSize > 0)

socket.SendBufferSize = Settings.SocketSendBufferSize;

if (Settings.TcpKeepAlive)

socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, true);

if (Settings.TcpKeepAliveInterval > 0 && Settings.TcpKeepAliveTime == 0)

throw new ArgumentException("If TcpKeepAliveInterval is defined, TcpKeepAliveTime must be defined as well");

if (Settings.TcpKeepAliveTime > 0)