<commit_before>﻿/*
Copyright 2017 Coin Foundry (coinfoundry.org)
Authors: <NAME> (<EMAIL>)

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and
associated documentation files (the "Software"), to deal in the Software without restriction,
including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense,
and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial
portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Security;
using System.Net.Sockets;
using System.Reactive;
using System.Runtime.InteropServices;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using System.Security.Policy;
using System.Threading;
using System.Threading.Tasks;
using Autofac;
using Microsoft.EntityFrameworkCore.Internal;
using MiningCore.Banning;
using MiningCore.Buffers;
using MiningCore.Configuration;
using MiningCore.JsonRpc;
using MiningCore.Time;
using MiningCore.Util;
using Newtonsoft.Json;
using NLog;
using Contract = MiningCore.Contracts.Contract;

namespace MiningCore.Stratum
{
    public abstract class StratumServer
    {
        protected StratumServer(IComponentContext ctx, IMasterClock clock)
        {
            Contract.RequiresNonNull(ctx, nameof(ctx));
            Contract.RequiresNonNull(clock, nameof(clock));

            this.ctx = ctx;
            this.clock = clock;
        }

        static StratumServer()
        {
            if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
            {
                ignoredSocketErrors = new HashSet<int>
                {
                    (int) SocketError.ConnectionReset,
                    (int) SocketError.ConnectionAborted,
                    (int) SocketError.OperationAborted
                };
            }

            else if (RuntimeInformation.IsOSPlatform(OSPlatform.Linux))
            {
                // see: http://www.virtsync.com/c-error-codes-include-errno
                ignoredSocketErrors = new HashSet<int>
                {
                    104,    // ECONNRESET
                    125,    // ECANCELED
                    103,    // ECONNABORTED
                    110,    // ETIMEDOUT
                };
            }
        }

        protected readonly Dictionary<string, StratumClient> clients = new Dictionary<string, StratumClient>();

        protected readonly IComponentContext ctx;
        protected readonly IMasterClock clock;
        protected readonly Dictionary<int, Socket> ports = new Dictionary<int, Socket>();
        protected ClusterConfig clusterConfig;
        protected IBanManager banManager;
        protected ILogger logger;

        protected static readonly HashSet<int> ignoredSocketErrors;

        protected abstract string LogCat { get; }

        public void StartListeners(params (IPEndPoint IPEndPoint, PoolEndpoint PoolEndpoint)[] stratumPorts)
        {
            Contract.RequiresNonNull(stratumPorts, nameof(stratumPorts));

            Task.Run(async () =>
            {
                // Setup sockets
                (Socket Socket, X509Certificate2 Cert)[] si = stratumPorts.Select(port =>
                {
                    // TLS cert loading
                    X509Certificate2 tlsCert = null;

                    if (port.PoolEndpoint.Tls)
                        tlsCert = new X509Certificate2(port.PoolEndpoint.TlsPfxFile);

                    // Setup socket
                    var server = new Socket(SocketType.Stream, ProtocolType.Tcp);
                    server.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
                    server.Bind(port.IPEndPoint);
                    server.Listen(512);

                    lock (ports)
                    {
                        ports[port.IPEndPoint.Port] = server;
                    }

                    logger.Info(() => $"[{LogCat}] Stratum port {port.IPEndPoint.Address}:{port.IPEndPoint.Port} online");

                    return (server, tlsCert);
                }).ToArray();

                // Setup accept tasks
                var tasks = si.Select(x => x.Socket.AcceptAsync()).ToArray();

                while (true)
                {
                    try
                    {
                        // Wait incoming connection on any of the monitored sockets
                        await Task.WhenAny(tasks);

                        // check tasks
                        for (var i = 0; i < tasks.Length; i++)
                        {
                            var task = tasks[i];
                            var port = stratumPorts[i];

                            // skip running tasks
                            if (!(task.IsCompleted || task.IsFaulted || task.IsCanceled))
                                continue;

                            // accept connection if successful
                            if (task.IsCompletedSuccessfully)
                                await AcceptConnectionAsync(task.Result, port, si[i].Cert);

                            // Refresh task
                            tasks[i] = si[i].Socket.AcceptAsync();
                        }
                    }

                    catch (Exception ex)
                    {
                        logger.Error(ex);
                    }
                }
            });
        }

        private async Task AcceptConnectionAsync(Socket socket, (IPEndPoint IPEndPoint, PoolEndpoint PoolEndpoint) port, X509Certificate2 tlsCert)
        {
            var remoteEndpoint = (IPEndPoint) socket.RemoteEndPoint;
            var connectionId = CorrelationIdGenerator.GetNextId();

            logger.Debug(() => $"[{LogCat}] Accepting connection [{connectionId}] from {remoteEndpoint.Address}:{remoteEndpoint.Port}");

            // get rid of banned clients as early as possible
            if (banManager?.IsBanned(remoteEndpoint.Address) == true)
            {
                logger.Debug(() => $"[{LogCat}] Disconnecting banned ip {remoteEndpoint.Address}");
                socket.Close();
                return;
            }

            // prepare socket
            socket.NoDelay = true;

            // create stream
            var stream = (Stream)new NetworkStream(socket, true);

            // TLS handshake
            if (port.PoolEndpoint.Tls)
            {
                SslStream sslStream = null;

                try
                {
                    sslStream = new SslStream(stream, false);
                    await sslStream.AuthenticateAsServerAsync(tlsCert, false, SslProtocols.Tls | SslProtocols.Tls11 | SslProtocols.Tls12, false);
                }

                catch (Exception ex)
                {
                    logger.Error(() => $"[{LogCat}] TLS init failed: {ex.Message}: {ex.InnerException.ToString() ?? string.Empty}");
                    (sslStream ?? stream).Close();
                    return;
                }

                stream = sslStream;
            }
            // setup client
            var client = new StratumClient(stream, clock, port.IPEndPoint, connectionId);

            lock (clients)
            {<commit_msg>TLS certificate sharing across pools<commit_after>﻿/*
Copyright 2017 Coin Foundry (coinfoundry.org)
Authors: <NAME> (<EMAIL>)

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and
associated documentation files (the "Software"), to deal in the Software without restriction,
including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense,
and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial
portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Security;
using System.Net.Sockets;
using System.Reactive;
using System.Runtime.InteropServices;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using System.Security.Policy;
using System.Threading;
using System.Threading.Tasks;
using Autofac;
using Microsoft.EntityFrameworkCore.Internal;
using MiningCore.Banning;
using MiningCore.Buffers;
using MiningCore.Configuration;
using MiningCore.JsonRpc;
using MiningCore.Time;
using MiningCore.Util;
using Newtonsoft.Json;
using NLog;
using Contract = MiningCore.Contracts.Contract;

namespace MiningCore.Stratum
{
    public abstract class StratumServer
    {
        protected StratumServer(IComponentContext ctx, IMasterClock clock)
        {
            Contract.RequiresNonNull(ctx, nameof(ctx));
            Contract.RequiresNonNull(clock, nameof(clock));

            this.ctx = ctx;
            this.clock = clock;
        }

        static StratumServer()
        {
            if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
            {
                ignoredSocketErrors = new HashSet<int>
                {
                    (int) SocketError.ConnectionReset,
                    (int) SocketError.ConnectionAborted,
                    (int) SocketError.OperationAborted
                };
            }

            else if (RuntimeInformation.IsOSPlatform(OSPlatform.Linux))
            {
                // see: http://www.virtsync.com/c-error-codes-include-errno
                ignoredSocketErrors = new HashSet<int>
                {
                    104,    // ECONNRESET
                    125,    // ECANCELED
                    103,    // ECONNABORTED
                    110,    // ETIMEDOUT
                };
            }
        }

        protected readonly Dictionary<string, StratumClient> clients = new Dictionary<string, StratumClient>();
        protected static readonly ConcurrentDictionary<string, X509Certificate2> certs = new ConcurrentDictionary<string, X509Certificate2>();

        protected readonly IComponentContext ctx;
        protected readonly IMasterClock clock;
        protected readonly Dictionary<int, Socket> ports = new Dictionary<int, Socket>();
        protected ClusterConfig clusterConfig;
        protected IBanManager banManager;
        protected ILogger logger;

        protected static readonly HashSet<int> ignoredSocketErrors;

        protected abstract string LogCat { get; }

        public void StartListeners(params (IPEndPoint IPEndPoint, PoolEndpoint PoolEndpoint)[] stratumPorts)
        {
            Contract.RequiresNonNull(stratumPorts, nameof(stratumPorts));

            Task.Run(async () =>
            {
                // Setup sockets
                var sockets = stratumPorts.Select(port =>
                {
                    // TLS cert loading
                    X509Certificate2 tlsCert = null;

                    if (port.PoolEndpoint.Tls)
                    {
                        if (!certs.TryGetValue(port.PoolEndpoint.TlsPfxFile, out tlsCert))
                        {
                            tlsCert = new X509Certificate2(port.PoolEndpoint.TlsPfxFile);
                            certs.TryAdd(port.PoolEndpoint.TlsPfxFile, tlsCert);
                        }
                    }

                    // Setup socket
                    var server = new Socket(SocketType.Stream, ProtocolType.Tcp);
                    server.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
                    server.Bind(port.IPEndPoint);
                    server.Listen(512);

                    lock (ports)
                    {
                        ports[port.IPEndPoint.Port] = server;
                    }

                    logger.Info(() => $"[{LogCat}] Stratum port {port.IPEndPoint.Address}:{port.IPEndPoint.Port} online");

                    return server;
                }).ToArray();

                // Setup accept tasks
                var tasks = sockets.Select(socket => socket.AcceptAsync()).ToArray();

                while (true)
                {
                    try
                    {
                        // Wait incoming connection on any of the monitored sockets
                        await Task.WhenAny(tasks);

                        // check tasks
                        for (var i = 0; i < tasks.Length; i++)
                        {
                            var task = tasks[i];
                            var port = stratumPorts[i];

                            // skip running tasks
                            if (!(task.IsCompleted || task.IsFaulted || task.IsCanceled))
                                continue;

                            // accept connection if successful
                            if (task.IsCompletedSuccessfully)
                                await AcceptConnectionAsync(task.Result, port);

                            // Refresh task
                            tasks[i] = sockets[i].AcceptAsync();
                        }
                    }

                    catch (Exception ex)
                    {
                        logger.Error(ex);
                    }
                }
            });
        }

        private async Task AcceptConnectionAsync(Socket socket, (IPEndPoint IPEndPoint, PoolEndpoint PoolEndpoint) port)
        {
            var remoteEndpoint = (IPEndPoint) socket.RemoteEndPoint;
            var connectionId = CorrelationIdGenerator.GetNextId();

            logger.Debug(() => $"[{LogCat}] Accepting connection [{connectionId}] from {remoteEndpoint.Address}:{remoteEndpoint.Port}");

            // get rid of banned clients as early as possible
            if (banManager?.IsBanned(remoteEndpoint.Address) == true)
            {
                logger.Debug(() => $"[{LogCat}] Disconnecting banned ip {remoteEndpoint.Address}");
                socket.Close();
                return;
            }

            // prepare socket
            socket.NoDelay = true;

            // create stream
            var stream = (Stream)new NetworkStream(socket, true);

            // TLS handshake
            if (port.PoolEndpoint.Tls)
            {
                SslStream sslStream = null;

                try
                {
                    var tlsCert = certs[port.PoolEndpoint.TlsPfxFile];
                    sslStream = new SslStream(stream, false);

                    await sslStream.AuthenticateAsServerAsync(tlsCert, false, SslProtocols.Tls | SslProtocols.Tls11 | SslProtocols.Tls12, false);
                }

                catch (Exception ex)
                {
                    logger.Error(() => $"[{LogCat}] TLS init failed: {ex.Message}: {ex.InnerException.ToString() ?? string.Empty}");
                    (sslStream ?? stream).Close();
                    return;
                }

                stream = sslStream;
            }
            // setup client
            var client = new StratumClient(stream, clock, port.IPEndPoint, connectionId);

            lock (clients)
            {
{'fraction_non_alphanumeric': 0.07950097847358122, 'fraction_numerical': 0.005259295499021526, 'mean_word_length': 1.6668297455968688, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2125161', 'n_tokens_mistral': 4122, 'n_tokens_neox': 3769, 'n_words': 1085}