wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2646 from etschannen/feature-boost-ssl 

switched SSL implementation to use boost ssl
This commit is contained in:
Evan Tschannen 2020-02-21 14:10:19 -08:00 committed by GitHub
parent 65cdd23d64 3354e6b278
commit 51d6f4a44e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
39 changed files with 1440 additions and 1128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CMakeLists.txt
View File

@ -174,9 +174,6 @@ set(SEED "0x${SEED_}" CACHE STRING "Random seed for testing")
################################################################################
include(CompileBoost)
if(WITH_TLS)
add_subdirectory(FDBLibTLS)
endif()
add_subdirectory(flow)
add_subdirectory(fdbrpc)
add_subdirectory(fdbclient)

11
Makefile
View File

@ -49,7 +49,7 @@ ifeq ($(PLATFORM),Linux)
CXXFLAGS += -std=c++17
BOOST_BASEDIR ?= /opt
TLS_LIBDIR ?= /usr/local/lib
TLS_LIBDIR ?= /usr/local/lib64
DLEXT := so
java_DLEXT := so
TARGET_LIBC_VERSION ?= 2.11
@ -65,7 +65,7 @@ else ifeq ($(PLATFORM),Darwin)
.LIBPATTERNS := lib%.dylib lib%.a
BOOST_BASEDIR ?= ${HOME}
TLS_LIBDIR ?= /usr/local/lib
TLS_LIBDIR ?= /usr/local/lib64
DLEXT := dylib
java_DLEXT := jnilib
else
@ -110,8 +110,8 @@ CFLAGS += -DTLS_DISABLED
FDB_TLS_LIB :=
TLS_LIBS :=
else
FDB_TLS_LIB := lib/libFDBLibTLS.a
TLS_LIBS += $(addprefix $(TLS_LIBDIR)/,libtls.a libssl.a libcrypto.a)
FDB_TLS_LIB :=
TLS_LIBS += $(addprefix $(TLS_LIBDIR)/,libssl.a libcrypto.a)
endif
CXXFLAGS += -Wno-deprecated -DBOOST_ERROR_CODE_HEADER_ONLY -DBOOST_SYSTEM_NO_DEPRECATED
@ -124,9 +124,6 @@ VPATH += $(addprefix :,$(filter-out lib,$(patsubst -L%,%,$(filter -L%,$(LDFLAGS)
CS_PROJECTS := flow/actorcompiler flow/coveragetool fdbclient/vexillographer
CPP_PROJECTS := flow fdbrpc fdbclient fdbbackup fdbserver fdbcli bindings/c bindings/java fdbmonitor bindings/flow/tester bindings/flow
ifndef TLS_DISABLED
CPP_PROJECTS += FDBLibTLS
endif
OTHER_PROJECTS := bindings/python bindings/ruby bindings/go
CS_MK_GENERATED := $(CS_PROJECTS:=/generated.mk)

7
bindings/c/fdb_c.cpp
View File

@ -108,7 +108,12 @@ fdb_error_t fdb_network_set_option( FDBNetworkOption option,
}
fdb_error_t fdb_setup_network_impl() {
CATCH_AND_RETURN( API->setupNetwork(); );
CATCH_AND_RETURN(
try {
API->setupNetwork();
} catch (boost::system::system_error& e) {
return error_code_tls_error;
} );
}
fdb_error_t fdb_setup_network_v13( const char* localAddress ) {

2
bindings/flow/fdb_flow.actor.cpp
View File

@ -82,7 +82,7 @@ void fdb_flow_test() {
fdb->setupNetwork();
startThread(networkThread, fdb);
g_network = newNet2( false );
g_network = newNet2(false);
openTraceFile(NetworkAddress(), 1000000, 1000000, ".");
systemMonitor();

1
bindings/flow/tester/local.mk
View File

@ -23,6 +23,7 @@
fdb_flow_tester_CFLAGS := -Ibindings/c $(fdbrpc_CFLAGS)
fdb_flow_tester_LDFLAGS := -Llib $(fdbrpc_LDFLAGS) -lfdb_c
fdb_flow_tester_LIBS := lib/libfdb_flow.a lib/libflow.a lib/libfdb_c.$(DLEXT)
fdb_flow_tester_STATIC_LIBS := $(TLS_LIBS)
fdb_flow_tester: lib/libfdb_c.$(DLEXT)
@mkdir -p bindings/flow/bin

2
build/docker-compose.yaml
View File

@ -2,7 +2,7 @@ version: "3"
services:
common: &common
image: foundationdb/foundationdb-build:0.1.9
image: foundationdb/foundationdb-build:0.1.11
build-setup: &build-setup
<<: *common

23
cmake/FDBComponents.cmake
View File

@ -9,21 +9,32 @@ if(USE_VALGRIND)
endif()
################################################################################
# LibreSSL
# SSL
################################################################################
set(DISABLE_TLS OFF CACHE BOOL "Don't try to find LibreSSL and always build without TLS support")
if(DISABLE_TLS)
set(WITH_TLS OFF)
else()
set(LIBRESSL_USE_STATIC_LIBS TRUE)
find_package(LibreSSL)
if(LibreSSL_FOUND)
set(OPENSSL_USE_STATIC_LIBS TRUE)
find_package(OpenSSL)
if(NOT OPENSSL_FOUND)
set(LIBRESSL_USE_STATIC_LIBS TRUE)
find_package(LibreSSL)
if (LIBRESSL_FOUND)
add_library(OpenSSL::SSL ALIAS LibreSSL)
endif()
endif()
if(OPENSSL_FOUND OR LIBRESSL_FOUND)
set(WITH_TLS ON)
add_compile_options(-DHAVE_OPENSSL)
else()
message(STATUS "LibreSSL NOT Found - Will compile without TLS Support")
message(STATUS "You can set LibreSSL_ROOT to the LibreSSL install directory to help cmake find it")
message(STATUS "Neither OpenSSL nor LibreSSL were found - Will compile without TLS Support")
message(STATUS "You can set OPENSSL_ROOT_DIR or LibreSSL_ROOT to the LibreSSL install directory to help cmake find it")
set(WITH_TLS OFF)
endif()
if(WIN32)
message(STATUS "TLS is temporarilty disabled on macOS while libressl -> openssl transition happens")
set(WITH_TLS OFF)
endif()
endif()

20
fdbbackup/backup.actor.cpp
View File

@ -38,7 +38,6 @@
#include "fdbclient/json_spirit/json_spirit_writer_template.h"
#include "fdbrpc/Platform.h"
#include "fdbrpc/TLSConnection.h"
#include <stdarg.h>
#include <stdio.h>
@ -3072,22 +3071,22 @@ int main(int argc, char* argv[]) {
blobCredentials.push_back(args->OptionArg());
break;
#ifndef TLS_DISABLED
case TLSOptions::OPT_TLS_PLUGIN:
case TLSParams::OPT_TLS_PLUGIN:
args->OptionArg();
break;
case TLSOptions::OPT_TLS_CERTIFICATES:
case TLSParams::OPT_TLS_CERTIFICATES:
tlsCertPath = args->OptionArg();
break;
case TLSOptions::OPT_TLS_PASSWORD:
case TLSParams::OPT_TLS_PASSWORD:
tlsPassword = args->OptionArg();
break;
case TLSOptions::OPT_TLS_CA_FILE:
case TLSParams::OPT_TLS_CA_FILE:
tlsCAPath = args->OptionArg();
break;
case TLSOptions::OPT_TLS_KEY:
case TLSParams::OPT_TLS_KEY:
tlsKeyPath = args->OptionArg();
break;
case TLSOptions::OPT_TLS_VERIFY_PEERS:
case TLSParams::OPT_TLS_VERIFY_PEERS:
tlsVerifyPeers = args->OptionArg();
break;
#endif
@ -3658,6 +3657,13 @@ int main(int argc, char* argv[]) {
} catch (Error& e) {
TraceEvent(SevError, "MainError").error(e);
status = FDB_EXIT_MAIN_ERROR;
} catch (boost::system::system_error& e) {
if (g_network) {
TraceEvent(SevError, "MainError").error(unknown_error()).detail("RootException", e.what());
} else {
fprintf(stderr, "ERROR: %s (%d)\n", e.what(), e.code().value());
}
status = FDB_EXIT_MAIN_EXCEPTION;
} catch (std::exception& e) {
TraceEvent(SevError, "MainError").error(unknown_error()).detail("RootException", e.what());
status = FDB_EXIT_MAIN_EXCEPTION;

18
fdbcli/fdbcli.actor.cpp
View File

@ -33,7 +33,6 @@
#include "flow/DeterministicRandom.h"
#include "flow/SignalSafeUnwind.h"
#include "fdbrpc/TLSConnection.h"
#include "fdbrpc/Platform.h"
#include "flow/SimpleOpt.h"
@ -2507,22 +2506,22 @@ struct CLIOptions {
#ifndef TLS_DISABLED
// TLS Options
case TLSOptions::OPT_TLS_PLUGIN:
case TLSParams::OPT_TLS_PLUGIN:
args.OptionArg();
break;
case TLSOptions::OPT_TLS_CERTIFICATES:
case TLSParams::OPT_TLS_CERTIFICATES:
tlsCertPath = args.OptionArg();
break;
case TLSOptions::OPT_TLS_CA_FILE:
case TLSParams::OPT_TLS_CA_FILE:
tlsCAPath = args.OptionArg();
break;
case TLSOptions::OPT_TLS_KEY:
case TLSParams::OPT_TLS_KEY:
tlsKeyPath = args.OptionArg();
break;
case TLSOptions::OPT_TLS_PASSWORD:
case TLSParams::OPT_TLS_PASSWORD:
tlsPassword = args.OptionArg();
break;
case TLSOptions::OPT_TLS_VERIFY_PEERS:
case TLSParams::OPT_TLS_VERIFY_PEERS:
tlsVerifyPeers = args.OptionArg();
break;
#endif
@ -2573,7 +2572,7 @@ ACTOR Future<Void> addInterface( std::map<Key,std::pair<Value,ClientLeaderRegInt
(*address_interface)[ip_port2] = std::make_pair(kv.value, leaderInterf);
}
}
when( wait(delay(1.0)) ) {}
when( wait(delay(CLIENT_KNOBS->CLI_CONNECT_TIMEOUT)) ) {}
}
return Void();
}
@ -3758,5 +3757,8 @@ int main(int argc, char **argv) {
} catch (Error& e) {
printf("ERROR: %s (%d)\n", e.what(), e.code());
return 1;
} catch (boost::system::system_error& e) {
printf("ERROR: %s (%d)\n", e.what(), e.code().value());
return 1;
}
}

7
fdbclient/Knobs.cpp
View File

@ -201,7 +201,8 @@ ClientKnobs::ClientKnobs(bool randomize) {
init( CONSISTENCY_CHECK_RATE_LIMIT_MAX, 50e6 ); // Limit in per sec
init( CONSISTENCY_CHECK_ONE_ROUND_TARGET_COMPLETION_TIME, 7 * 24 * 60 * 60 ); // 7 days
//fdbcli
init( CLI_CONNECT_PARALLELISM, 10 );
//fdbcli
init( CLI_CONNECT_PARALLELISM, 400 );
init( CLI_CONNECT_TIMEOUT, 10.0 );
}

7
fdbclient/Knobs.h
View File

@ -191,10 +191,11 @@ public:
int CONSISTENCY_CHECK_RATE_LIMIT_MAX;
int CONSISTENCY_CHECK_ONE_ROUND_TARGET_COMPLETION_TIME;
//fdbcli
int CLI_CONNECT_PARALLELISM;
// fdbcli
int CLI_CONNECT_PARALLELISM;
double CLI_CONNECT_TIMEOUT;
ClientKnobs(bool randomize = false);
};

23
fdbclient/ManagementAPI.actor.cpp
View File

@ -982,9 +982,13 @@ ACTOR Future<CoordinatorsResult::Type> changeQuorum( Database cx, Reference<IQuo
if(g_network->isSimulated()) {
for(int i = 0; i < (desiredCoordinators.size()/2)+1; i++) {
auto address = NetworkAddress(desiredCoordinators[i].ip,desiredCoordinators[i].port,true,false);
g_simulator.protectedAddresses.insert(address);
TraceEvent("ProtectCoordinator").detail("Address", address).backtrace();
auto addresses = g_simulator.getProcessByAddress(desiredCoordinators[i])->addresses;
g_simulator.protectedAddresses.insert(addresses.address);
if(addresses.secondaryAddress.present()) {
g_simulator.protectedAddresses.insert(addresses.secondaryAddress.get());
}
TraceEvent("ProtectCoordinator").detail("Address", desiredCoordinators[i]).backtrace();
}
}
@ -1143,8 +1147,7 @@ struct AutoQuorumChange : IQuorumChange {
*err = CoordinatorsResult::NOT_ENOUGH_MACHINES;
return vector<NetworkAddress>();
}
desiredCount = std::max(oldCoordinators.size(), (workers.size() - 1) | 1);
chosen.resize(desiredCount);
chosen.resize((chosen.size() - 1) | 1);
}
return chosen;
@ -1518,10 +1521,14 @@ ACTOR Future<std::set<NetworkAddress>> checkForExcludingServers(Database cx, vec
state bool ok = true;
inProgressExclusion.clear();
for(auto& s : serverList) {
auto addr = decodeServerListValue( s.value ).address();
if ( addressExcluded(exclusions, addr) ) {
auto addresses = decodeServerListValue( s.value ).getKeyValues.getEndpoint().addresses;
if ( addressExcluded(exclusions, addresses.address) ) {
ok = false;
inProgressExclusion.insert(addr);
inProgressExclusion.insert(addresses.address);
}
if ( addresses.secondaryAddress.present() && addressExcluded(exclusions, addresses.secondaryAddress.get()) ) {
ok = false;
inProgressExclusion.insert(addresses.secondaryAddress.get());
}
}

70
fdbclient/NativeAPI.actor.cpp
View File

@ -38,12 +38,12 @@
#include "fdbrpc/LoadBalance.h"
#include "fdbrpc/Net2FileSystem.h"
#include "fdbrpc/simulator.h"
#include "fdbrpc/TLSConnection.h"
#include "flow/ActorCollection.h"
#include "flow/DeterministicRandom.h"
#include "flow/Knobs.h"
#include "flow/Platform.h"
#include "flow/SystemMonitor.h"
#include "flow/TLSPolicy.h"
#include "flow/UnitTest.h"
#if defined(CMAKE_BUILD) || !defined(WIN32)
@ -67,12 +67,15 @@ using std::min;
using std::pair;
NetworkOptions networkOptions;
Reference<TLSOptions> tlsOptions;
TLSParams tlsParams;
static Reference<TLSPolicy> tlsPolicy;
static void initTLSOptions() {
if (!tlsOptions) {
tlsOptions = Reference<TLSOptions>(new TLSOptions());
static void initTLSPolicy() {
#ifndef TLS_DISABLED
if (!tlsPolicy) {
tlsPolicy = Reference<TLSPolicy>(new TLSPolicy(TLSPolicy::Is::CLIENT));
}
#endif
}
static const Key CLIENT_LATENCY_INFO_PREFIX = LiteralStringRef("client_latency/");
@ -887,49 +890,46 @@ void setNetworkOption(FDBNetworkOptions::Option option, Optional<StringRef> valu
break;
case FDBNetworkOptions::TLS_CERT_PATH:
validateOptionValue(value, true);
initTLSOptions();
tlsOptions->set_cert_file( value.get().toString() );
tlsParams.tlsCertPath = value.get().toString();
break;
case FDBNetworkOptions::TLS_CERT_BYTES:
initTLSOptions();
tlsOptions->set_cert_data( value.get().toString() );
break;
case FDBNetworkOptions::TLS_CA_PATH:
case FDBNetworkOptions::TLS_CERT_BYTES: {
validateOptionValue(value, true);
initTLSOptions();
tlsOptions->set_ca_file( value.get().toString() );
tlsParams.tlsCertBytes = value.get().toString();
break;
case FDBNetworkOptions::TLS_CA_BYTES:
}
case FDBNetworkOptions::TLS_CA_PATH: {
validateOptionValue(value, true);
initTLSOptions();
tlsOptions->set_ca_data(value.get().toString());
tlsParams.tlsCAPath = value.get().toString();
break;
}
case FDBNetworkOptions::TLS_CA_BYTES: {
validateOptionValue(value, true);
tlsParams.tlsCABytes = value.get().toString();
break;
}
case FDBNetworkOptions::TLS_PASSWORD:
validateOptionValue(value, true);
initTLSOptions();
tlsOptions->set_key_password(value.get().toString());
tlsParams.tlsPassword = value.get().toString();
break;
case FDBNetworkOptions::TLS_KEY_PATH:
validateOptionValue(value, true);
initTLSOptions();
tlsOptions->set_key_file( value.get().toString() );
validateOptionValue(value, true);
tlsParams.tlsKeyPath = value.get().toString();
break;
case FDBNetworkOptions::TLS_KEY_BYTES:
case FDBNetworkOptions::TLS_KEY_BYTES: {
validateOptionValue(value, true);
initTLSOptions();
tlsOptions->set_key_data( value.get().toString() );
tlsParams.tlsKeyBytes = value.get().toString();
break;
}
case FDBNetworkOptions::TLS_VERIFY_PEERS:
validateOptionValue(value, true);
initTLSOptions();
try {
tlsOptions->set_verify_peers({ value.get().toString() });
} catch( Error& e ) {
initTLSPolicy();
#ifndef TLS_DISABLED
if (!tlsPolicy->set_verify_peers({ value.get().toString() })) {
TraceEvent(SevWarnAlways, "TLSValidationSetError")
.error( e )
.detail("Input", value.get().toString() );
throw invalid_option_value();
}
#endif
break;
case FDBNetworkOptions::CLIENT_BUGGIFY_ENABLE:
enableBuggify(true, BuggifyType::Client);
@ -987,15 +987,11 @@ void setupNetwork(uint64_t transportId, bool useMetrics) {
if (!networkOptions.logClientInfo.present())
networkOptions.logClientInfo = true;
g_network = newNet2(false, useMetrics || networkOptions.traceDirectory.present());
initTLSPolicy();
g_network = newNet2(false, useMetrics || networkOptions.traceDirectory.present(), tlsPolicy, tlsParams);
FlowTransport::createInstance(true, transportId);
Net2FileSystem::newFileSystem();
initTLSOptions();
#ifndef TLS_DISABLED
tlsOptions->register_network();
#endif
}
void runNetwork() {

2
fdbclient/md5/md5.c
View File

@ -35,7 +35,7 @@
* compile-time configuration.
*/
#ifndef HAVE_OPENSSL
#if !defined(HAVE_OPENSSL) || defined(TLS_DISABLED)
#include <string.h>

2
fdbclient/md5/md5.h
View File

@ -23,7 +23,7 @@
* See md5.c for more information.
*/
#ifdef HAVE_OPENSSL
#if defined(HAVE_OPENSSL) && !defined(TLS_DISABLED)
#include <openssl/md5.h>
#elif !defined(_MD5_H)
#define _MD5_H

1
fdbrpc/CMakeLists.txt
View File

@ -27,7 +27,6 @@ set(FDBRPC_SRCS
sim2.actor.cpp
sim_validation.cpp
TimedRequest.h
TLSConnection.actor.cpp
TraceFileIO.cpp)
set(FDBRPC_THIRD_PARTY_SRCS

25
fdbrpc/FlowTransport.actor.cpp
View File

@ -302,7 +302,7 @@ ACTOR Future<Void> connectionMonitor( Reference<Peer> peer ) {
state double lastRefreshed = now();
state int64_t lastBytesReceived = peer->bytesReceived;
loop {
wait(delay(FLOW_KNOBS->CONNECTION_MONITOR_LOOP_TIME));
wait(delay(FLOW_KNOBS->CONNECTION_MONITOR_LOOP_TIME, TaskPriority::ReadSocket));
if (lastBytesReceived < peer->bytesReceived) {
lastRefreshed = now();
lastBytesReceived = peer->bytesReceived;
@ -317,7 +317,7 @@ ACTOR Future<Void> connectionMonitor( Reference<Peer> peer ) {
//We cannot let an error be thrown from connectionMonitor while still on the stack from scanPackets in connectionReader
//because then it would not call the destructor of connectionReader when connectionReader is cancelled.
wait(delay(0));
wait(delay(0, TaskPriority::ReadSocket));
if (peer->reliable.empty() && peer->unsent.empty() && peer->outstandingReplies==0) {
if (peer->peerReferences == 0 &&
@ -332,7 +332,7 @@ ACTOR Future<Void> connectionMonitor( Reference<Peer> peer ) {
}
}
wait (delayJittered(FLOW_KNOBS->CONNECTION_MONITOR_LOOP_TIME));
wait (delayJittered(FLOW_KNOBS->CONNECTION_MONITOR_LOOP_TIME, TaskPriority::ReadSocket));
// TODO: Stop monitoring and close the connection with no onDisconnect requests outstanding
state ReplyPromise<Void> reply;
@ -432,15 +432,17 @@ ACTOR Future<Void> connectionKeeper( Reference<Peer> self,
try {
choose {
when( Reference<IConnection> _conn = wait( INetworkConnections::net()->connect(self->destination) ) ) {
conn = _conn;
wait(conn->connectHandshake());
if (FlowTransport::transport().isClient()) {
IFailureMonitor::failureMonitor().setStatus(self->destination, FailureStatus(false));
}
if (self->unsent.empty()) {
_conn->close();
conn->close();
conn = Reference<IConnection>();
clientReconnectDelay = false;
continue;
} else {
conn = _conn;
TraceEvent("ConnectionExchangingConnectPacket", conn->getDebugID())
.suppressFor(1.0)
.detail("PeerAddr", self->destination);
@ -962,6 +964,7 @@ ACTOR static Future<Void> connectionReader(
ACTOR static Future<Void> connectionIncoming( TransportData* self, Reference<IConnection> conn ) {
try {
wait(conn->acceptHandshake());
state Promise<Reference<Peer>> onConnected;
state Future<Void> reader = connectionReader( self, conn, Reference<Peer>(), onConnected );
choose {
@ -988,11 +991,13 @@ ACTOR static Future<Void> listen( TransportData* self, NetworkAddress listenAddr
try {
loop {
Reference<IConnection> conn = wait( listener->accept() );
TraceEvent("ConnectionFrom", conn->getDebugID()).suppressFor(1.0)
.detail("FromAddress", conn->getPeerAddress())
.detail("ListenAddress", listenAddr.toString());
incoming.add( connectionIncoming(self, conn) );
wait(delay(0) || delay(FLOW_KNOBS->CONNECTION_ACCEPT_DELAY, TaskPriority::WriteSocket));
if(conn) {
TraceEvent("ConnectionFrom", conn->getDebugID()).suppressFor(1.0)
.detail("FromAddress", conn->getPeerAddress())
.detail("ListenAddress", listenAddr.toString());
incoming.add( connectionIncoming(self, conn) );
}
wait(delay(0, TaskPriority::AcceptSocket));
}
} catch (Error& e) {
TraceEvent(SevError, "ListenError").error(e);

136
fdbrpc/ITLSPlugin.h
View File

@ -1,136 +0,0 @@
/*
* ITLSPlugin.h
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2018 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#ifndef FDB_ITLSPLUGIN_H
#define FDB_ITLSPLUGIN_H
#pragma once
#include <stdint.h>
struct ITLSSession {
enum { SUCCESS = 0, WANT_READ = -1, WANT_WRITE = -2, FAILED = -3 };
virtual void addref() = 0;
virtual void delref() = 0;
// handshake should return SUCCESS if the handshake is complete,
// FAILED on fatal error, or one of WANT_READ or WANT_WRITE if the
// handshake should be reattempted after more data can be
// read/written on the underlying connection.
virtual int handshake() = 0;
// read should return the (non-zero) number of bytes read,
// WANT_READ or WANT_WRITE if the operation is blocked by the
// underlying stream, or FAILED if there is an error (including a
// closed connection).
virtual int read(uint8_t* data, int length) = 0;
// write should return the (non-zero) number of bytes written, or
// WANT_READ or WANT_WRITE if the operation is blocked by the
// underlying stream, or FAILED if there is an error.
virtual int write(const uint8_t* data, int length) = 0;
};
// Returns the number of bytes sent (possibly 0), or -1 on error
// (including connection close)
typedef int(*TLSSendCallbackFunc)(void* ctx, const uint8_t* buf, int len);
// Returns the number of bytes read (possibly 0), or -1 on error
// (including connection close)
typedef int(*TLSRecvCallbackFunc)(void* ctx, uint8_t* buf, int len);
struct ITLSPolicy {
virtual void addref() = 0;
virtual void delref() = 0;
// set_ca_data should import the provided certificate list and
// associate it with this policy. cert_data will point to a PEM
// encoded certificate list of trust roots.
//
// set_ca_data should return true if the operation succeeded,
// and false otherwise. After the first call to create_session for
// a given policy, set_ca_data should immediately return false
// if called.
virtual bool set_ca_data(const uint8_t* ca_data, int ca_len) = 0;
// set_cert_data should import the provided certificate list and
// associate it with this policy. cert_data will point to a PEM
// encoded certificate list, ordered such that each certificate
// certifies the one before it.
//
// cert_data may additionally contain key information, which must
// be ignored.
//
// set_cert_data should return true if the operation succeeded,
// and false otherwise. After the first call to create_session for
// a given policy, set_cert_data should immediately return false
// if called.
virtual bool set_cert_data(const uint8_t* cert_data, int cert_len) = 0;
// set_key_data should import the provided private key and
// associate it with this policy. key_data will point to a PEM
// encoded key, which may be encrypted. If encrypted the password
// argument should be specified, otherwise it may be NULL.
//
// key_data may additionally contain certificate information,
// which must be ignored.
//
// set_key_data should return true if the operation succeeded, and
// false otherwise. After the first call to create_session for a
// given policy, set_key_data should immediately return false if
// called.
virtual bool set_key_data(const uint8_t* key_data, int key_len, const char* password) = 0;
// set_verify_peers should modify the validation rules for
// verifying a peer during connection handshake. The format of
// verify_peers is implementation specific.
//
// set_verify_peers should return true if the operation succeed,
// and false otherwise. After the first call to create_session for
// a given policy, set_verify_peers should immediately return
// false if called.
virtual bool set_verify_peers(int count, const uint8_t* verify_peers[], int verify_peers_len[]) = 0;
// create_session should return a new object that implements
// ITLSSession, associated with this policy. After the first call
// to create_session for a given policy, further calls to
// ITLSPolicy::set_* will fail and return false.
//
// The newly created session should use send_func and recv_func to
// send and receive data on the underlying transport, and must
// provide send_ctx/recv_ctx to the callbacks.
//
// uid will be used to identify this session within trace events
virtual ITLSSession* create_session(bool is_client, const char *servername, TLSSendCallbackFunc send_func, void* send_ctx, TLSRecvCallbackFunc recv_func, void* recv_ctx, void* uid) = 0;
};
struct ITLSPlugin {
virtual void addref() = 0;
virtual void delref() = 0;
// create_policy should return a new object that implements
// ITLSPolicy.
virtual ITLSPolicy* create_policy() = 0;
static inline const char* get_plugin_type_name_and_version() { return "ITLSPlugin"; }
};
#endif /* FDB_ITLSPLUGIN_H */

26
fdbrpc/LoadPlugin.h
View File

@ -18,27 +18,21 @@
* limitations under the License.
*/
#ifndef _FLOW_LOADPLUGIN_H_
#define _FLOW_LOADPLUGIN_H_
#pragma once
// Specialized TLS plugin library
extern "C" void *get_tls_plugin(const char *plugin_type_name_and_version);
// Name of specialized TLS Plugin
extern const char* tlsPluginName;
#include <string>
#include "flow/flow.h"
template <class T>
Reference<T> loadPlugin( std::string const& plugin_name ) {
void *(*get_plugin)(const char*) = NULL;
#ifndef TLS_DISABLED
if (!plugin_name.compare(tlsPluginName)) {
get_plugin = (void*(*)(const char*)) get_tls_plugin;
}
else
#endif
{
void* plugin = loadLibrary( plugin_name.c_str() );
if (plugin)
get_plugin = (void*(*)(const char*))loadFunction( plugin, "get_plugin" );
}
void* plugin = loadLibrary( plugin_name.c_str() );
if (plugin)
get_plugin = (void*(*)(const char*))loadFunction( plugin, "get_plugin" );
return (get_plugin) ? Reference<T>( (T*)get_plugin( T::get_plugin_type_name_and_version() ) ) : Reference<T>( NULL );
}
#endif

545
fdbrpc/TLSConnection.actor.cpp
View File

@ -1,545 +0,0 @@
/*
* TLSConnection.actor.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2018 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <memory>
#include "flow/flow.h"
#include "flow/network.h"
#include "flow/Knobs.h"
#include "fdbrpc/TLSConnection.h"
#include "fdbrpc/ITLSPlugin.h"
#include "fdbrpc/LoadPlugin.h"
#include "fdbrpc/Platform.h"
#include "fdbrpc/IAsyncFile.h"
#include "flow/actorcompiler.h" // This must be the last #include.
// Name of specialized TLS Plugin
const char* tlsPluginName = "fdb-libressl-plugin";
// Must not throw an exception from this function!
static int send_func(void* ctx, const uint8_t* buf, int len) {
TLSConnection* conn = (TLSConnection*)ctx;
try {
SendBuffer sb;
sb.bytes_sent = 0;
sb.bytes_written = len;
sb.data = buf;
sb.next = 0;
int w = conn->conn->write( &sb );
return w;
} catch ( Error& e ) {
TraceEvent("TLSConnectionSendError", conn->getDebugID()).suppressFor(1.0).detail("Peer", conn->getPeerAddress().toString()).error(e);
return -1;
} catch ( ... ) {
TraceEvent("TLSConnectionSendError", conn->getDebugID()).suppressFor(1.0).detail("Peer", conn->getPeerAddress()).error( unknown_error() );
return -1;
}
}
// Must not throw an exception from this function!
static int recv_func(void* ctx, uint8_t* buf, int len) {
TLSConnection* conn = (TLSConnection*)ctx;
try {
int r = conn->conn->read( buf, buf + len );
return r;
} catch ( Error& e ) {
TraceEvent("TLSConnectionRecvError", conn->getDebugID()).suppressFor(1.0).detail("Peer", conn->getPeerAddress()).error(e);
return -1;
} catch ( ... ) {
TraceEvent("TLSConnectionRecvError", conn->getDebugID()).suppressFor(1.0).detail("Peer", conn->getPeerAddress()).error( unknown_error() );
return -1;
}
}
ACTOR static Future<Void> handshake( TLSConnection* self ) {
state std::pair<IPAddress,uint16_t> peerIP = std::make_pair(self->conn->getPeerAddress().ip, self->is_client ? self->conn->getPeerAddress().port : static_cast<uint16_t>(0));
if(!self->is_client) {
auto iter(g_network->networkInfo.serverTLSConnectionThrottler.find(peerIP));
if(iter != g_network->networkInfo.serverTLSConnectionThrottler.end()) {
if (now() < iter->second.second) {
if(iter->second.first >= FLOW_KNOBS->TLS_SERVER_CONNECTION_THROTTLE_ATTEMPTS) {
TraceEvent("TLSIncomingConnectionThrottlingWarning", self->getDebugID()).suppressFor(1.0).detail("PeerIP", peerIP.first.toString());
wait(delay(FLOW_KNOBS->CONNECTION_MONITOR_TIMEOUT));
throw connection_failed();
}
} else {
g_network->networkInfo.serverTLSConnectionThrottler.erase(peerIP);
}
}
}
loop {
int r = self->session->handshake();
if(BUGGIFY_WITH_PROB(0.001)) {
r = ITLSSession::FAILED;
}
if ( r == ITLSSession::SUCCESS ) break;
if ( r == ITLSSession::FAILED ) {
TraceEvent("TLSConnectionHandshakeError", self->getDebugID()).suppressFor(1.0).detail("Peer", self->getPeerAddress());
auto iter(g_network->networkInfo.serverTLSConnectionThrottler.find(peerIP));
if(iter != g_network->networkInfo.serverTLSConnectionThrottler.end()) {
iter->second.first++;
} else {
g_network->networkInfo.serverTLSConnectionThrottler[peerIP] = std::make_pair(0,now() + (self->is_client ? FLOW_KNOBS->TLS_CLIENT_CONNECTION_THROTTLE_TIMEOUT : FLOW_KNOBS->TLS_SERVER_CONNECTION_THROTTLE_TIMEOUT));
}
throw connection_failed();
}
ASSERT( r == ITLSSession::WANT_WRITE || r == ITLSSession::WANT_READ );
wait( r == ITLSSession::WANT_WRITE ? self->conn->onWritable() : self->conn->onReadable() );
}
TraceEvent("TLSConnectionHandshakeSuccessful", self->getDebugID()).suppressFor(1.0).detail("Peer", self->getPeerAddress());
return Void();
}
TLSConnection::TLSConnection( Reference<IConnection> const& conn, Reference<ITLSPolicy> const& policy, bool is_client, std::string host) : conn(conn), write_wants(0), read_wants(0), uid(conn->getDebugID()), is_client(is_client) {
const char * serverName = host.empty() ? NULL : host.c_str();
session = Reference<ITLSSession>( policy->create_session(is_client, serverName, send_func, this, recv_func, this, (void*)&uid) );
if ( !session ) {
// If session is NULL, we're trusting policy->create_session
// to have used its provided logging function to have logged
// the error
throw tls_error();
}
handshook = handshake(this);
}
Future<Void> TLSConnection::onWritable() {
if ( !handshook.isReady() )
return handshook;
return
write_wants == ITLSSession::WANT_READ ? conn->onReadable() :
write_wants == ITLSSession::WANT_WRITE ? conn->onWritable() :
Void();
}
Future<Void> TLSConnection::onReadable() {
if ( !handshook.isReady() )
return handshook;
return
read_wants == ITLSSession::WANT_READ ? conn->onReadable() :
read_wants == ITLSSession::WANT_WRITE ? conn->onWritable() :
Void();
}
int TLSConnection::read( uint8_t* begin, uint8_t* end ) {
if ( !handshook.isReady() ) return 0;
handshook.get();
read_wants = 0;
int r = session->read( begin, end - begin );
if ( r > 0 )
return r;
if ( r == ITLSSession::FAILED ) throw connection_failed();
ASSERT( r == ITLSSession::WANT_WRITE || r == ITLSSession::WANT_READ );
read_wants = r;
return 0;
}
int TLSConnection::write( SendBuffer const* buffer, int limit ) {
ASSERT(limit > 0);
if ( !handshook.isReady() ) return 0;
handshook.get();
write_wants = 0;
int toSend = std::min(limit, buffer->bytes_written - buffer->bytes_sent);
ASSERT(toSend);
int w = session->write( buffer->data + buffer->bytes_sent, toSend );
if ( w > 0 )
return w;
if ( w == ITLSSession::FAILED ) throw connection_failed();
ASSERT( w == ITLSSession::WANT_WRITE || w == ITLSSession::WANT_READ );
write_wants = w;
return 0;
}
ACTOR Future<Reference<IConnection>> wrap( Reference<ITLSPolicy> policy, bool is_client, Future<Reference<IConnection>> c, std::string host) {
state Reference<IConnection> conn = wait(c);
try {
state Reference<TLSConnection> tlsConn(new TLSConnection( conn, policy, is_client, host ));
if(is_client) {
wait(tlsConn->handshook);
}
return tlsConn;
} catch( Error &e ) {
conn->close();
throw e;
}
}
Future<Reference<IConnection>> TLSListener::accept() {
return wrap( options->get_policy(TLSOptions::POLICY_VERIFY_PEERS), false, listener->accept(), "");
}
TLSNetworkConnections::TLSNetworkConnections( Reference<TLSOptions> options ) : options(options) {
network = INetworkConnections::net();
g_network->setGlobal(INetwork::enumGlobal::enNetworkConnections, (flowGlobalType) this);
}
ACTOR Future<Reference<IConnection>> waitAndFailConnection() {
wait(delay(FLOW_KNOBS->CONNECTION_MONITOR_TIMEOUT));
throw connection_failed();
}
Future<Reference<IConnection>> TLSNetworkConnections::connect( NetworkAddress toAddr, std::string host) {
if ( toAddr.isTLS() ) {
NetworkAddress clearAddr( toAddr.ip, toAddr.port, toAddr.isPublic(), false );
std::pair<IPAddress,uint16_t> peerIP = std::make_pair(toAddr.ip, toAddr.port);
auto iter(g_network->networkInfo.serverTLSConnectionThrottler.find(peerIP));
if(iter != g_network->networkInfo.serverTLSConnectionThrottler.end()) {
if (now() < iter->second.second) {
if(iter->second.first >= FLOW_KNOBS->TLS_CLIENT_CONNECTION_THROTTLE_ATTEMPTS) {
TraceEvent("TLSOutgoingConnectionThrottlingWarning").suppressFor(1.0).detail("PeerIP", toAddr);
return waitAndFailConnection();
}
} else {
g_network->networkInfo.serverTLSConnectionThrottler.erase(peerIP);
}
}
TraceEvent("TLSConnectionConnecting").suppressFor(1.0).detail("ToAddr", toAddr);
// For FDB<->FDB connections, we don't have hostnames and can't verify IP
// addresses against certificates, so we have our own peer verifying logic
// to use. For FDB<->external system connections, we can use the standard
// hostname-based certificate verification logic.
if (host.empty() || host == toAddr.ip.toString())
return wrap(options->get_policy(TLSOptions::POLICY_VERIFY_PEERS), true, network->connect(clearAddr), std::string(""));
else
return wrap( options->get_policy(TLSOptions::POLICY_NO_VERIFY_PEERS), true, network->connect( clearAddr ), host );
}
return network->connect( toAddr );
}
Future<std::vector<NetworkAddress>> TLSNetworkConnections::resolveTCPEndpoint( std::string host, std::string service) {
return network->resolveTCPEndpoint( host, service );
}
Reference<IListener> TLSNetworkConnections::listen( NetworkAddress localAddr ) {
if ( localAddr.isTLS() ) {
NetworkAddress clearAddr( localAddr.ip, localAddr.port, localAddr.isPublic(), false );
TraceEvent("TLSConnectionListening").detail("OnAddr", localAddr);
return Reference<IListener>(new TLSListener( options, network->listen( clearAddr ) ));
}
return network->listen( localAddr );
}
// 5MB for loading files into memory
#define CERT_FILE_MAX_SIZE (5 * 1024 * 1024)
void TLSOptions::set_cert_file( std::string const& cert_file ) {
try {
TraceEvent("TLSConnectionSettingCertFile").detail("CertFilePath", cert_file);
policyInfo.cert_path = cert_file;
set_cert_data( readFileBytes( cert_file, CERT_FILE_MAX_SIZE ) );
} catch ( Error& e) {
TraceEvent(SevError, "TLSOptionsSetCertFileError").detail("Filename", cert_file).error(e).GetLastError();
throw;
}
}
void TLSOptions::set_ca_file(std::string const& ca_file) {
try {
TraceEvent("TLSConnectionSettingCAFile").detail("CAPath", ca_file);
policyInfo.ca_path = ca_file;
set_ca_data(readFileBytes(ca_file, CERT_FILE_MAX_SIZE));
}
catch (Error& e) {
TraceEvent(SevError, "TLSOptionsSetCertAError").detail("Filename", ca_file).error(e).GetLastError();
throw;
}
}
void TLSOptions::set_ca_data(std::string const& ca_data) {
if (!policyVerifyPeersSet.get() || !policyVerifyPeersNotSet.get())
init_plugin();
TraceEvent("TLSConnectionSettingCAData").detail("CADataSize", ca_data.size());
policyInfo.ca_contents = Standalone<StringRef>(ca_data);
if (!policyVerifyPeersSet.get()->set_ca_data((const uint8_t*)&ca_data[0], ca_data.size()))
throw tls_error();
if (!policyVerifyPeersNotSet.get()->set_ca_data((const uint8_t*)&ca_data[0], ca_data.size()))
throw tls_error();
ca_set = true;
}
void TLSOptions::set_cert_data( std::string const& cert_data ) {
if (!policyVerifyPeersSet.get() || !policyVerifyPeersNotSet.get())
init_plugin();
TraceEvent("TLSConnectionSettingCertData").detail("CertDataSize", cert_data.size());
policyInfo.cert_contents = Standalone<StringRef>(cert_data);
if ( !policyVerifyPeersSet.get()->set_cert_data( (const uint8_t*)&cert_data[0], cert_data.size() ) )
throw tls_error();
if (!policyVerifyPeersNotSet.get()->set_cert_data((const uint8_t*)&cert_data[0], cert_data.size()))
throw tls_error();
certs_set = true;
}
void TLSOptions::set_key_password(std::string const& password) {
TraceEvent("TLSConnectionSettingPassword");
policyInfo.keyPassword = password;
}
void TLSOptions::set_key_file( std::string const& key_file ) {
try {
TraceEvent("TLSConnectionSettingKeyFile").detail("KeyFilePath", key_file);
policyInfo.key_path = key_file;
set_key_data( readFileBytes( key_file, CERT_FILE_MAX_SIZE ) );
} catch ( Error& e) {
TraceEvent(SevError, "TLSOptionsSetKeyFileError").detail("Filename", key_file).error(e).GetLastError();
throw;
}
}
void TLSOptions::set_key_data( std::string const& key_data ) {
if (!policyVerifyPeersSet.get() || !policyVerifyPeersNotSet.get())
init_plugin();
const char *passphrase = policyInfo.keyPassword.empty() ? NULL : policyInfo.keyPassword.c_str();
TraceEvent("TLSConnectionSettingKeyData").detail("KeyDataSize", key_data.size());
policyInfo.key_contents = Standalone<StringRef>(key_data);
if ( !policyVerifyPeersSet.get()->set_key_data( (const uint8_t*)&key_data[0], key_data.size(), passphrase) )
throw tls_error();
if (!policyVerifyPeersNotSet.get()->set_key_data((const uint8_t*)&key_data[0], key_data.size(), passphrase))
throw tls_error();
key_set = true;
}
void TLSOptions::set_verify_peers( std::vector<std::string> const& verify_peers ) {
if (!policyVerifyPeersSet.get())
init_plugin();
{
TraceEvent e("TLSConnectionSettingVerifyPeers");
for (int i = 0; i < verify_peers.size(); i++)
e.detail(std::string("Value" + std::to_string(i)).c_str(), verify_peers[i].c_str());
}
std::unique_ptr<const uint8_t *[]> verify_peers_arr(new const uint8_t*[verify_peers.size()]);
std::unique_ptr<int[]> verify_peers_len(new int[verify_peers.size()]);
for (int i = 0; i < verify_peers.size(); i++) {
verify_peers_arr[i] = (const uint8_t *)&verify_peers[i][0];
verify_peers_len[i] = verify_peers[i].size();
}
if (!policyVerifyPeersSet.get()->set_verify_peers(verify_peers.size(), verify_peers_arr.get(), verify_peers_len.get()))
throw tls_error();
policyInfo.verify_peers = verify_peers;
verify_peers_set = true;
}
void TLSOptions::register_network() {
// Simulation relies upon being able to call this multiple times, and have it override g_network
// each time it's called.
new TLSNetworkConnections( Reference<TLSOptions>::addRef( this ) );
}
ACTOR static Future<ErrorOr<Standalone<StringRef>>> readEntireFile( std::string filename ) {
state Reference<IAsyncFile> file = wait(IAsyncFileSystem::filesystem()->open(filename, IAsyncFile::OPEN_READONLY | IAsyncFile::OPEN_UNCACHED, 0));
state int64_t filesize = wait(file->size());
state Standalone<StringRef> buf = makeString(filesize);
int rc = wait(file->read(mutateString(buf), filesize, 0));
if (rc != filesize) {
// File modified during read, probably. The mtime should change, and thus we'll be called again.
return tls_error();
}
return buf;
}
ACTOR static Future<Void> watchFileForChanges( std::string filename, AsyncVar<Standalone<StringRef>> *contents_var ) {
state std::time_t lastModTime = wait(IAsyncFileSystem::filesystem()->lastWriteTime(filename));
loop {
wait(delay(FLOW_KNOBS->TLS_CERT_REFRESH_DELAY_SECONDS));
std::time_t modtime = wait(IAsyncFileSystem::filesystem()->lastWriteTime(filename));
if (lastModTime != modtime) {
lastModTime = modtime;
ErrorOr<Standalone<StringRef>> contents = wait(readEntireFile(filename));
if (contents.present()) {
contents_var->set(contents.get());
}
}
}
}
ACTOR static Future<Void> reloadConfigurationOnChange( TLSOptions::PolicyInfo *pci, Reference<ITLSPlugin> plugin, AsyncVar<Reference<ITLSPolicy>> *realVerifyPeersPolicy, AsyncVar<Reference<ITLSPolicy>> *realNoVerifyPeersPolicy ) {
if (FLOW_KNOBS->TLS_CERT_REFRESH_DELAY_SECONDS <= 0) {
return Void();
}
loop {
// Early in bootup, the filesystem might not be initialized yet. Wait until it is.
if (IAsyncFileSystem::filesystem() != nullptr) {
break;
}
wait(delay(1.0));
}
state int mismatches = 0;
state AsyncVar<Standalone<StringRef>> ca_var;
state AsyncVar<Standalone<StringRef>> key_var;
state AsyncVar<Standalone<StringRef>> cert_var;
state std::vector<Future<Void>> lifetimes;
if (!pci->ca_path.empty()) lifetimes.push_back(watchFileForChanges(pci->ca_path, &ca_var));
if (!pci->key_path.empty()) lifetimes.push_back(watchFileForChanges(pci->key_path, &key_var));
if (!pci->cert_path.empty()) lifetimes.push_back(watchFileForChanges(pci->cert_path, &cert_var));
loop {
state Future<Void> ca_changed = ca_var.onChange();
state Future<Void> key_changed = key_var.onChange();
state Future<Void> cert_changed = cert_var.onChange();
wait( ca_changed || key_changed || cert_changed );
if (ca_changed.isReady()) {
TraceEvent(SevInfo, "TLSRefreshCAChanged").detail("path", pci->ca_path).detail("length", ca_var.get().size());
pci->ca_contents = ca_var.get();
}
if (key_changed.isReady()) {
TraceEvent(SevInfo, "TLSRefreshKeyChanged").detail("path", pci->key_path).detail("length", key_var.get().size());
pci->key_contents = key_var.get();
}
if (cert_changed.isReady()) {
TraceEvent(SevInfo, "TLSRefreshCertChanged").detail("path", pci->cert_path).detail("length", cert_var.get().size());
pci->cert_contents = cert_var.get();
}
bool rc = true;
Reference<ITLSPolicy> verifypeers = Reference<ITLSPolicy>(plugin->create_policy());
Reference<ITLSPolicy> noverifypeers = Reference<ITLSPolicy>(plugin->create_policy());
loop {
// Don't actually loop. We're just using loop/break as a `goto err`.
// This loop always ends with an unconditional break.
rc = verifypeers->set_ca_data(pci->ca_contents.begin(), pci->ca_contents.size());
if (!rc) break;
rc = verifypeers->set_key_data(pci->key_contents.begin(), pci->key_contents.size(), pci->keyPassword.c_str());
if (!rc) break;
rc = verifypeers->set_cert_data(pci->cert_contents.begin(), pci->cert_contents.size());
if (!rc) break;
{
std::unique_ptr<const uint8_t *[]> verify_peers_arr(new const uint8_t*[pci->verify_peers.size()]);
std::unique_ptr<int[]> verify_peers_len(new int[pci->verify_peers.size()]);
for (int i = 0; i < pci->verify_peers.size(); i++) {
verify_peers_arr[i] = (const uint8_t *)&pci->verify_peers[i][0];
verify_peers_len[i] = pci->verify_peers[i].size();
}
rc = verifypeers->set_verify_peers(pci->verify_peers.size(), verify_peers_arr.get(), verify_peers_len.get());
if (!rc) break;
}
rc = noverifypeers->set_ca_data(pci->ca_contents.begin(), pci->ca_contents.size());
if (!rc) break;
rc = noverifypeers->set_key_data(pci->key_contents.begin(), pci->key_contents.size(), pci->keyPassword.c_str());
if (!rc) break;
rc = noverifypeers->set_cert_data(pci->cert_contents.begin(), pci->cert_contents.size());
if (!rc) break;
break;
}
if (rc) {
TraceEvent(SevInfo, "TLSCertificateRefreshSucceeded");
realVerifyPeersPolicy->set(verifypeers);
realNoVerifyPeersPolicy->set(noverifypeers);
mismatches = 0;
} else {
// Some files didn't match up, they should in the future, and we'll retry then.
mismatches++;
TraceEvent(SevWarn, "TLSCertificateRefreshMismatch").detail("mismatches", mismatches);
}
}
}
const char *defaultCertFileName = "fdb.pem";
Reference<ITLSPolicy> TLSOptions::get_policy(PolicyType type) {
if ( !certs_set ) {
if ( !platform::getEnvironmentVar( "FDB_TLS_CERTIFICATE_FILE", policyInfo.cert_path ) )
policyInfo.cert_path = fileExists(defaultCertFileName) ? defaultCertFileName : joinPath(platform::getDefaultConfigPath(), defaultCertFileName);
set_cert_file( policyInfo.cert_path );
}
if ( !key_set ) {
if ( policyInfo.keyPassword.empty() )
platform::getEnvironmentVar( "FDB_TLS_PASSWORD", policyInfo.keyPassword );
if ( !platform::getEnvironmentVar( "FDB_TLS_KEY_FILE", policyInfo.key_path ) )
policyInfo.key_path = fileExists(defaultCertFileName) ? defaultCertFileName : joinPath(platform::getDefaultConfigPath(), defaultCertFileName);
set_key_file( policyInfo.key_path );
}
if( !verify_peers_set ) {
std::string verify_peers;
if (platform::getEnvironmentVar("FDB_TLS_VERIFY_PEERS", verify_peers))
set_verify_peers({ verify_peers });
else
set_verify_peers({ std::string("Check.Valid=1")});
}
if (!ca_set) {
if (platform::getEnvironmentVar("FDB_TLS_CA_FILE", policyInfo.ca_path))
set_ca_file(policyInfo.ca_path);
}
if (!configurationReloader.present()) {
configurationReloader = reloadConfigurationOnChange(&policyInfo, plugin, &policyVerifyPeersSet, &policyVerifyPeersNotSet);
}
Reference<ITLSPolicy> policy;
switch (type) {
case POLICY_VERIFY_PEERS:
policy = policyVerifyPeersSet.get();
break;
case POLICY_NO_VERIFY_PEERS:
policy = policyVerifyPeersNotSet.get();
break;
default:
ASSERT_ABORT(0);
}
return policy;
}
void TLSOptions::init_plugin() {
TraceEvent("TLSConnectionLoadingPlugin").detail("Plugin", tlsPluginName);
plugin = loadPlugin<ITLSPlugin>( tlsPluginName );
if ( !plugin ) {
TraceEvent(SevError, "TLSConnectionPluginInitError").detail("Plugin", tlsPluginName).GetLastError();
throw tls_error();
}
policyVerifyPeersSet = AsyncVar<Reference<ITLSPolicy>>(Reference<ITLSPolicy>(plugin->create_policy()));
if ( !policyVerifyPeersSet.get()) {
// Hopefully create_policy logged something with the log func
TraceEvent(SevError, "TLSConnectionCreatePolicyVerifyPeersSetError");
throw tls_error();
}
policyVerifyPeersNotSet = AsyncVar<Reference<ITLSPolicy>>(Reference<ITLSPolicy>(plugin->create_policy()));
if (!policyVerifyPeersNotSet.get()) {
// Hopefully create_policy logged something with the log func
TraceEvent(SevError, "TLSConnectionCreatePolicyVerifyPeersNotSetError");
throw tls_error();
}
}
bool TLSOptions::enabled() {
return policyVerifyPeersSet.get().isValid() && policyVerifyPeersNotSet.get().isValid();
}

174
fdbrpc/TLSConnection.h
View File

@ -1,174 +0,0 @@
/*
* TLSConnection.h
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2018 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#ifndef FLOW_TLSCONNECTION_H
#define FLOW_TLSCONNECTION_H
#pragma once
#include "flow/Platform.h"
#include "fdbrpc/ITLSPlugin.h"
struct TLSConnection : IConnection, ReferenceCounted<TLSConnection> {
Reference<IConnection> conn;
Reference<ITLSSession> session;
Future<Void> handshook;
int write_wants, read_wants;
UID uid;
bool is_client;
virtual void addref() { ReferenceCounted<TLSConnection>::addref(); }
virtual void delref() { ReferenceCounted<TLSConnection>::delref(); }
TLSConnection( Reference<IConnection> const& conn, Reference<ITLSPolicy> const& policy, bool is_client, std::string host);
~TLSConnection() {
// Here for ordering to make sure we delref the ITLSSession
// which has a pointer to this object
session.clear();
}
virtual void close() { conn->close(); }
virtual Future<Void> onWritable();
virtual Future<Void> onReadable();
virtual int read( uint8_t* begin, uint8_t* end );
virtual int write( SendBuffer const* buffer, int limit);
virtual NetworkAddress getPeerAddress() {
NetworkAddress a = conn->getPeerAddress();
return NetworkAddress(a.ip, a.port, a.isPublic(), true);
}
virtual UID getDebugID() { return uid; }
};
struct TLSOptions : ReferenceCounted<TLSOptions> {
enum { OPT_TLS = 100000, OPT_TLS_PLUGIN, OPT_TLS_CERTIFICATES, OPT_TLS_KEY, OPT_TLS_VERIFY_PEERS, OPT_TLS_CA_FILE, OPT_TLS_PASSWORD };
enum PolicyType { POLICY_VERIFY_PEERS = 1, POLICY_NO_VERIFY_PEERS };
TLSOptions() : certs_set(false), key_set(false), verify_peers_set(false), ca_set(false) {
#ifndef TLS_DISABLED
init_plugin( );
#endif
}
void set_cert_file( std::string const& cert_file );
void set_cert_data( std::string const& cert_data );
void set_ca_file(std::string const& ca_file);
void set_ca_data(std::string const& ca_data);
// If there is a passphrase, this api should be called prior to setting key for the passphrase to be used
void set_key_password( std::string const& password );
void set_key_file( std::string const& key_file );
void set_key_data( std::string const& key_data );
void set_verify_peers( std::vector<std::string> const& verify_peers );
void register_network();
Reference<ITLSPolicy> get_policy(PolicyType type);
bool enabled();
struct PolicyInfo {
std::string ca_path;
Standalone<StringRef> ca_contents;
std::string key_path;
std::string keyPassword;
Standalone<StringRef> key_contents;
std::string cert_path;
Standalone<StringRef> cert_contents;
std::vector<std::string> verify_peers;
};
private:
void init_plugin();
Reference<ITLSPlugin> plugin;
PolicyInfo policyInfo;
AsyncVar<Reference<ITLSPolicy>> policyVerifyPeersSet;
AsyncVar<Reference<ITLSPolicy>> policyVerifyPeersNotSet;
Optional<Future<Void>> configurationReloader;
bool certs_set, key_set, verify_peers_set, ca_set;
};
struct TLSListener : IListener, ReferenceCounted<TLSListener> {
Reference<IListener> listener;
Reference<TLSOptions> options;
TLSListener( Reference<TLSOptions> options, Reference<IListener> listener ) : options(options), listener(listener) {}
virtual void addref() { ReferenceCounted<TLSListener>::addref(); }
virtual void delref() { ReferenceCounted<TLSListener>::delref(); }
virtual Future<Reference<IConnection>> accept();
virtual NetworkAddress getListenAddress() { return listener->getListenAddress(); }
};
struct TLSNetworkConnections : INetworkConnections {
INetworkConnections *network;
explicit TLSNetworkConnections( Reference<TLSOptions> options );
virtual Future<Reference<IConnection>> connect( NetworkAddress toAddr, std::string host );
virtual Future<std::vector<NetworkAddress>> resolveTCPEndpoint( std::string host, std::string service);
virtual Reference<IListener> listen( NetworkAddress localAddr );
private:
Reference<TLSOptions> options;
};
#define TLS_PLUGIN_FLAG "--tls_plugin"
#define TLS_CERTIFICATE_FILE_FLAG "--tls_certificate_file"
#define TLS_KEY_FILE_FLAG "--tls_key_file"
#define TLS_VERIFY_PEERS_FLAG "--tls_verify_peers"
#define TLS_CA_FILE_FLAG "--tls_ca_file"
#define TLS_PASSWORD_FLAG "--tls_password"
#define TLS_OPTION_FLAGS \
{ TLSOptions::OPT_TLS_PLUGIN, TLS_PLUGIN_FLAG, SO_REQ_SEP }, \
{ TLSOptions::OPT_TLS_CERTIFICATES, TLS_CERTIFICATE_FILE_FLAG, SO_REQ_SEP }, \
{ TLSOptions::OPT_TLS_KEY, TLS_KEY_FILE_FLAG, SO_REQ_SEP }, \
{ TLSOptions::OPT_TLS_VERIFY_PEERS, TLS_VERIFY_PEERS_FLAG, SO_REQ_SEP }, \
{ TLSOptions::OPT_TLS_PASSWORD, TLS_PASSWORD_FLAG, SO_REQ_SEP }, \
{ TLSOptions::OPT_TLS_CA_FILE, TLS_CA_FILE_FLAG, SO_REQ_SEP },
#define TLS_HELP \
" " TLS_CERTIFICATE_FILE_FLAG " CERTFILE\n" \
" The path of a file containing the TLS certificate and CA\n" \
" chain.\n" \
" " TLS_CA_FILE_FLAG " CERTAUTHFILE\n" \
" The path of a file containing the CA certificates chain.\n" \
" " TLS_KEY_FILE_FLAG " KEYFILE\n" \
" The path of a file containing the private key corresponding\n" \
" to the TLS certificate.\n" \
" " TLS_PASSWORD_FLAG " PASSCODE\n" \
" The passphrase of encrypted private key\n" \
" " TLS_VERIFY_PEERS_FLAG " CONSTRAINTS\n" \
" The constraints by which to validate TLS peers. The contents\n" \
" and format of CONSTRAINTS are plugin-specific.\n"
#endif /* FLOW_TLSCONNECTION_H */

3
fdbrpc/fdbrpc.vcxproj
View File

@ -35,7 +35,6 @@
<ClCompile Include="ReplicationTypes.cpp" />
<ClCompile Include="ReplicationPolicy.cpp" />
<ClCompile Include="sim_validation.cpp" />
<ActorCompiler Include="TLSConnection.actor.cpp" />
<ClCompile Include="TraceFileIO.cpp" />
<ClCompile Include="zlib\gzwrite.c" />
<ClCompile Include="zlib\gzclose.c" />
@ -91,7 +90,6 @@
<ClInclude Include="Platform.h" />
<ClInclude Include="fdbrpc.h" />
<ClInclude Include="FlowTransport.h" />
<ClInclude Include="ITLSPlugin.h" />
<ClInclude Include="libcoroutine\Base.h" />
<ClInclude Include="libcoroutine\Common.h" />
<ClInclude Include="libcoroutine\Coro.h" />
@ -110,7 +108,6 @@
<ClInclude Include="sim_validation.h" />
<ClInclude Include="Smoother.h" />
<ClInclude Include="TimedRequest.h" />
<ClInclude Include="TLSConnection.h" />
<ClInclude Include="TraceFileIO.h" />
<ClInclude Include="zlib\zlib.h" />
<ClInclude Include="zlib\deflate.h" />

3
fdbrpc/fdbrpc.vcxproj.filters
View File

@ -10,7 +10,6 @@
<ActorCompiler Include="AsyncFileCached.actor.cpp" />
<ActorCompiler Include="AsyncFileNonDurable.actor.h" />
<ActorCompiler Include="AsyncFileNonDurable.actor.cpp" />
<ActorCompiler Include="TLSConnection.actor.cpp" />
<ActorCompiler Include="dsltest.actor.cpp" />
<ActorCompiler Include="FlowTests.actor.cpp" />
<ActorCompiler Include="genericactors.actor.cpp" />
@ -129,7 +128,6 @@
<ClInclude Include="zlib\inftrees.h">
<Filter>zlib</Filter>
</ClInclude>
<ClInclude Include="ITLSPlugin.h" />
<ClInclude Include="FailureMonitor.h" />
<ClInclude Include="FlowTransport.h" />
<ClInclude Include="IAsyncFile.h" />
@ -144,7 +142,6 @@
<ClInclude Include="RangeMap.h" />
<ClInclude Include="Smoother.h" />
<ClInclude Include="TraceFileIO.h" />
<ClInclude Include="TLSConnection.h" />
<ClInclude Include="IRateControl.h" />
<ClInclude Include="Replication.h" />
<ClInclude Include="ReplicationTypes.h" />

16
fdbrpc/sim2.actor.cpp
View File

@ -200,6 +200,9 @@ struct Sim2Conn : IConnection, ReferenceCounted<Sim2Conn> {
virtual void delref() { ReferenceCounted<Sim2Conn>::delref(); }
virtual void close() { closedByCaller = true; closeInternal(); }
virtual Future<Void> acceptHandshake() { return delay(0.01*deterministicRandom()->random01()); }
virtual Future<Void> connectHandshake() { return delay(0.01*deterministicRandom()->random01()); }
virtual Future<Void> onWritable() { return whenWritable(this); }
virtual Future<Void> onReadable() { return whenReadable(this); }
@ -807,7 +810,7 @@ public:
}
// Sets the taskID/priority of the current task, without yielding
virtual Future<Reference<IConnection>> connect( NetworkAddress toAddr, std::string host ) {
ASSERT( !toAddr.isTLS() && host.empty());
ASSERT( host.empty());
if (!addressMap.count( toAddr )) {
return waitForProcessAndConnect( toAddr, this );
}
@ -825,7 +828,7 @@ public:
} else {
localIp = IPAddress(getCurrentProcess()->address.ip.toV4() + deterministicRandom()->randomInt(0, 256));
}
peerc->connect(myc, NetworkAddress(localIp, deterministicRandom()->randomInt(40000, 60000)));
peerc->connect(myc, NetworkAddress(localIp, deterministicRandom()->randomInt(40000, 60000), false, toAddr.isTLS()));
((Sim2Listener*)peerp->getListener(toAddr).getPtr())->incomingConnection( 0.5*deterministicRandom()->random01(), Reference<IConnection>(peerc) );
return onConnect( ::delay(0.5*deterministicRandom()->random01()), myc );
@ -846,7 +849,6 @@ public:
return conn;
}
virtual Reference<IListener> listen( NetworkAddress localAddr ) {
ASSERT( !localAddr.isTLS() );
Reference<IListener> listener( getCurrentProcess()->getListener(localAddr) );
ASSERT(listener);
return listener;
@ -1001,7 +1003,7 @@ public:
virtual void run() {
_run(this);
}
virtual ProcessInfo* newProcess(const char* name, IPAddress ip, uint16_t port, uint16_t listenPerProcess,
virtual ProcessInfo* newProcess(const char* name, IPAddress ip, uint16_t port, bool sslEnabled, uint16_t listenPerProcess,
LocalityData locality, ProcessClass startingClass, const char* dataFolder,
const char* coordinationFolder) {
ASSERT( locality.machineId().present() );
@ -1030,14 +1032,14 @@ public:
}
NetworkAddressList addresses;
addresses.address = NetworkAddress(ip, port, true, false);
addresses.address = NetworkAddress(ip, port, true, sslEnabled);
if(listenPerProcess == 2) {
addresses.secondaryAddress = NetworkAddress(ip, port+1, true, false);
}
ProcessInfo* m = new ProcessInfo(name, locality, startingClass, addresses, this, dataFolder, coordinationFolder);
for (int processPort = port; processPort < port + listenPerProcess; ++processPort) {
NetworkAddress address(ip, processPort, true, false); // SOMEDAY see above about becoming SSL!
NetworkAddress address(ip, processPort, true, sslEnabled && processPort == port);
m->listenerMap[address] = Reference<IListener>( new Sim2Listener(m, address) );
addressMap[address] = m;
}
@ -1570,7 +1572,7 @@ public:
return processes;
}
virtual ProcessInfo* getProcessByAddress( NetworkAddress const& address ) {
NetworkAddress normalizedAddress(address.ip, address.port, true, false);
NetworkAddress normalizedAddress(address.ip, address.port, true, address.isTLS());
ASSERT( addressMap.count( normalizedAddress ) );
return addressMap[ normalizedAddress ];
}

2
fdbrpc/simulator.h
View File

@ -140,7 +140,7 @@ public:
virtual Future<Void> onProcess( ISimulator::ProcessInfo *process, TaskPriority taskID = TaskPriority::Zero ) = 0;
virtual Future<Void> onMachine( ISimulator::ProcessInfo *process, TaskPriority taskID = TaskPriority::Zero ) = 0;
virtual ProcessInfo* newProcess(const char* name, IPAddress ip, uint16_t port, uint16_t listenPerProcess,
virtual ProcessInfo* newProcess(const char* name, IPAddress ip, uint16_t port, bool sslEnabled, uint16_t listenPerProcess,
LocalityData locality, ProcessClass startingClass, const char* dataFolder,
const char* coordinationFolder) = 0;
virtual void killProcess( ProcessInfo* machine, KillType ) = 0;

107
fdbserver/SimulatedCluster.actor.cpp
View File

@ -30,7 +30,6 @@
#include "fdbserver/CoordinationInterface.h"
#include "fdbmonitor/SimpleIni.h"
#include "fdbrpc/AsyncFileNonDurable.actor.h"
#include "fdbrpc/TLSConnection.h"
#include "fdbclient/ManagementAPI.actor.h"
#include "fdbclient/NativeAPI.actor.h"
#include "fdbclient/BackupAgent.actor.h"
@ -49,60 +48,6 @@ const int MACHINE_REBOOT_TIME = 10;
bool destructed = false;
static const char* certBytes =
"-----BEGIN CERTIFICATE-----\n"
"MIIEGzCCAwOgAwIBAgIJANUQj1rRA2XMMA0GCSqGSIb3DQEBBQUAMIGjMQswCQYD\n"
"VQQGEwJVUzELMAkGA1UECAwCVkExDzANBgNVBAcMBlZpZW5uYTEaMBgGA1UECgwR\n"
"Rm91bmRhdGlvbkRCLCBMTEMxGTAXBgNVBAsMEFRlc3QgZW5naW5lZXJpbmcxFTAT\n"
"BgNVBAMMDE1yLiBCaWcgVHVuYTEoMCYGCSqGSIb3DQEJARYZYmlnLnR1bmFAZm91\n"
"bmRhdGlvbmRiLmNvbTAeFw0xNDEyMDUxNTEyMjFaFw0yNDEyMDIxNTEyMjFaMIGj\n"
"MQswCQYDVQQGEwJVUzELMAkGA1UECAwCVkExDzANBgNVBAcMBlZpZW5uYTEaMBgG\n"
"A1UECgwRRm91bmRhdGlvbkRCLCBMTEMxGTAXBgNVBAsMEFRlc3QgZW5naW5lZXJp\n"
"bmcxFTATBgNVBAMMDE1yLiBCaWcgVHVuYTEoMCYGCSqGSIb3DQEJARYZYmlnLnR1\n"
"bmFAZm91bmRhdGlvbmRiLmNvbTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoC\n"
"ggEBAKZTL2edDkiet4HBTZnjysn6gOVZH2MP02KVBIv/H7e+3w7ZOIRvcPzhZe9M\n"
"3cGH1t/pkr9DSXvzIb42EffMVlpLD2VQn2H8VC2QSdJCIQcf802u+Taf+XtW6K1h\n"
"p/YPL1uhdopUs3c1oon8ykKwnOfrQYgv5pUa7jQdMkltI2MQJU3uFq3Z/LHTvIKe\n"
"FN+bqK0iYhZthwMG7Rld4+RgKZoT4u1B6w/duEWk9KLjgs7fTf3Oe6JHCYNqwBJi\n"
"78sJalwXz9Wf8wmMaYSG0XNA7vBOdpTFhVPSsh6e3rkydf5HydMade/II98MWpMe\n"
"hFg7FFMaJP6ig8p5iL+9QP2VMCkCAwEAAaNQME4wHQYDVR0OBBYEFIXGmIcKptBP\n"
"v3i9WS/mK78o5E/MMB8GA1UdIwQYMBaAFIXGmIcKptBPv3i9WS/mK78o5E/MMAwG\n"
"A1UdEwQFMAMBAf8wDQYJKoZIhvcNAQEFBQADggEBAJkVgNGOXT+ZHCNEYLjr/6OM\n"
"UCHvwlMeaEyqxaOmK26J2kAADPhjBZ7lZOHWb2Wzb+BiQUIFGwNIMoRvsg8skpJa\n"
"OCqpVciHVXY/U8BiYY70DKozRza93Ab9om3pySGDJ/akdCjqbMT1Cb7Kloyw+hNh\n"
"XD4MML0lYiUE9KK35xyK6FgTx4A7IXl4b3lWBgglqTh4+P5J1+xy8AYJ0VfPoP7y\n"
"OoZgwAmkpkMnalReNkN7LALHGqMzv/qH04ODlkU/HUGgExtnINMxK9VEDIe/yLGm\n"
"DHy7gcQMj5Hyymack/d4ZF8CSrYpGZQeZGXoxOmTDwWcXgnYA+2o7lOYPb5Uu08=\n"
"-----END CERTIFICATE-----\n"
"-----BEGIN PRIVATE KEY-----\n"
"MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQCmUy9nnQ5InreB\n"
"wU2Z48rJ+oDlWR9jD9NilQSL/x+3vt8O2TiEb3D84WXvTN3Bh9bf6ZK/Q0l78yG+\n"
"NhH3zFZaSw9lUJ9h/FQtkEnSQiEHH/NNrvk2n/l7VuitYaf2Dy9boXaKVLN3NaKJ\n"
"/MpCsJzn60GIL+aVGu40HTJJbSNjECVN7hat2fyx07yCnhTfm6itImIWbYcDBu0Z\n"
"XePkYCmaE+LtQesP3bhFpPSi44LO3039znuiRwmDasASYu/LCWpcF8/Vn/MJjGmE\n"
"htFzQO7wTnaUxYVT0rIent65MnX+R8nTGnXvyCPfDFqTHoRYOxRTGiT+ooPKeYi/\n"
"vUD9lTApAgMBAAECggEBAIYCmDtfq9aPK0P8v82yX/4FPD2OZV+nrKXNc3BpCuE9\n"
"hPOtyX/LWrol0b/Rqwr3rAWVaIt6Z4bbCuD7J9cEaL8voyP6pbCJYjmj/BbQ+VOI\n"
"Rrzcsid1Fcpu5+JqwK3c5kdp/NzQChmOuXt8lmrNal7iilZ0YdDZdfu/WnkW2mBB\n"
"oQHkujlnWr4PNYdwMOnBU6TwdOuz+inPVMLohOO0Vr585OxPsGzG2Ud3yQ/t34Cq\n"
"F9nmOXQoszftGKsL1yuh/3fGj/O86g/CRsUy05qZhDDBEYQD6qZCvD5+yp8oOWIR\n"
"SljM3GXDBnJqRPhP+Nyf6e6/GoQtfVZ9MPRzDDPzIBECgYEA2kX/zAs6taOiNqCb\n"
"6nVGe7/3uQJz/CkmOSKIFKUu7lCEUjmMYpK3Xzp26RTUR9cT+g9y+cnJO1Vbaxtf\n"
"Qidje6K+Oi1pQyUGQ6W+U8cPJHz43PVa7IB5Az5i/sS2tu0BGhvGo9G6iYQjxXeD\n"
"1197DRACgnm5AORQMum616XvSPMCgYEAwxKbkAzJzfZF6A3Ys+/0kycNfDP8xZoC\n"
"1zV3d1b2JncsdAPCHYSKtpniRrQN9ASa3RMdkh+wrMN/KlbtU9Ddoc4NHxSTFV7F\n"
"wypFMzLZslqkQ6uHnVVewHV7prfoKsMci2c9iHO7W8TEv4aqW8XDd8OozP3/q2j4\n"
"hvL7VIAVqXMCgYEAwAFnfOQ75uBkp00tGlfDgsRhc5vWz3CbMRNRRWfxGq41V+dL\n"
"uMJ7EAfr5ijue6uU5RmF+HkqzUjOvC894oGnn3CPibm8qNX+5q7799JZXa2ZdTVX\n"
"oEd7LAFLL/V3DP77Qy4/1Id/Ycydcu0pSuGw6tK0gnX06fXtHnxAYcaT8UUCgYAE\n"
"MytcP5o8r/ezVlD7Fsh6PpYAvZHMo1M6VPFchWfJTjmLyeTtA8SEx+1iPlAql8rJ\n"
"xbaWRc5k+dSMEdEMQ+vxpuELcUL1a9PwLsHMp2SefWsZ9eB2l7bxh9YAsebyvL6p\n"
"lbBydqNrB2KBCSIz1Z8uveytdS6C/0CSjzqwCA3vVwKBgQDAXqjo3xrzMlHeXm5o\n"
"qH/OjajjqbnPXHolHDitbLubyQ4E6KhMBMxfChBe/8VptB/Gs0efVbMVGuabxY7Q\n"
"iastGId8HyONy3UPGPxCn4b95cIxKvdpt+hvWtYHIBCfHXluQK7zsDMgvtXjYNiz\n"
"peZRikYlwmu1K2YRTf7oLE2Ogw==\n"
"-----END PRIVATE KEY-----\n";
template <class T>
T simulate( const T& in ) {
BinaryWriter writer(AssumeVersion(currentProtocolVersion));
@ -113,13 +58,6 @@ T simulate( const T& in ) {
return out;
}
static void simInitTLS(Reference<TLSOptions> tlsOptions) {
tlsOptions->set_cert_data( certBytes );
tlsOptions->set_key_data( certBytes );
tlsOptions->set_verify_peers(std::vector<std::string>(1, "Check.Valid=0"));
tlsOptions->register_network();
}
ACTOR Future<Void> runBackup( Reference<ClusterConnectionFile> connFile ) {
state std::vector<Future<Void>> agentFutures;
@ -196,7 +134,7 @@ enum AgentMode {
// a loop{} will be needed around the waiting on simulatedFDBD(). For now this simply
// takes care of house-keeping such as context switching and file closing.
ACTOR Future<ISimulator::KillType> simulatedFDBDRebooter(Reference<ClusterConnectionFile> connFile, IPAddress ip,
bool sslEnabled, Reference<TLSOptions> tlsOptions,
bool sslEnabled,
uint16_t port, uint16_t listenPerProcess,
LocalityData localities, ProcessClass processClass,
std::string* dataFolder, std::string* coordFolder,
@ -218,7 +156,7 @@ ACTOR Future<ISimulator::KillType> simulatedFDBDRebooter(Reference<ClusterConnec
wait( delay( waitTime ) );
state ISimulator::ProcessInfo* process =
g_simulator.newProcess("Server", ip, port, listenPerProcess, localities, processClass, dataFolder->c_str(),
g_simulator.newProcess("Server", ip, port, sslEnabled, listenPerProcess, localities, processClass, dataFolder->c_str(),
coordFolder->c_str());
wait(g_simulator.onProcess(process,
TaskPriority::DefaultYield)); // Now switch execution to the process on which we will run
@ -247,9 +185,6 @@ ACTOR Future<ISimulator::KillType> simulatedFDBDRebooter(Reference<ClusterConnec
//SOMEDAY: test lower memory limits, without making them too small and causing the database to stop making progress
FlowTransport::createInstance(processClass == ProcessClass::TesterClass || runBackupAgents == AgentOnly, 1);
Sim2FileSystem::newFileSystem();
if (sslEnabled) {
tlsOptions->register_network();
}
vector<Future<Void>> futures;
for (int listenPort = port; listenPort < port + listenPerProcess; ++listenPort) {
@ -363,8 +298,7 @@ std::string describe(int const& val) {
// Since a datacenter kill is considered to be the same as killing a machine, files cannot be swapped across datacenters
std::map< Optional<Standalone<StringRef>>, std::vector< std::vector< std::string > > > availableFolders;
// process count is no longer needed because it is now the length of the vector of ip's, because it was one ip per process
ACTOR Future<Void> simulatedMachine(ClusterConnectionString connStr, std::vector<IPAddress> ips, bool sslEnabled,
Reference<TLSOptions> tlsOptions, LocalityData localities,
ACTOR Future<Void> simulatedMachine(ClusterConnectionString connStr, std::vector<IPAddress> ips, bool sslEnabled, LocalityData localities,
ProcessClass processClass, std::string baseFolder, bool restarting,
bool useSeedFile, AgentMode runBackupAgents, bool sslOnly, std::string whitelistBinPaths) {
state int bootCount = 0;
@ -409,7 +343,7 @@ ACTOR Future<Void> simulatedMachine(ClusterConnectionString connStr, std::vector
Reference<ClusterConnectionFile> clusterFile(useSeedFile ? new ClusterConnectionFile(path, connStr.toString()) : new ClusterConnectionFile(path));
const int listenPort = i*listenPerProcess + 1;
AgentMode agentMode = runBackupAgents == AgentOnly ? ( i == ips.size()-1 ? AgentOnly : AgentNone ) : runBackupAgents;
processes.push_back(simulatedFDBDRebooter(clusterFile, ips[i], sslEnabled, tlsOptions, listenPort, listenPerProcess, localities, processClass, &myFolders[i], &coordFolders[i], baseFolder, connStr, useSeedFile, agentMode, whitelistBinPaths));
processes.push_back(simulatedFDBDRebooter(clusterFile, ips[i], sslEnabled, listenPort, listenPerProcess, localities, processClass, &myFolders[i], &coordFolders[i], baseFolder, connStr, useSeedFile, agentMode, whitelistBinPaths));
TraceEvent("SimulatedMachineProcess", randomId).detail("Address", NetworkAddress(ips[i], listenPort, true, false)).detail("ZoneId", localities.zoneId()).detail("DataHall", localities.dataHallId()).detail("Folder", myFolders[i]);
}
@ -614,7 +548,7 @@ IPAddress makeIPAddressForSim(bool isIPv6, std::array<int, 4> parts) {
ACTOR Future<Void> restartSimulatedSystem(vector<Future<Void>>* systemActors, std::string baseFolder, int* pTesterCount,
Optional<ClusterConnectionString>* pConnString,
Standalone<StringRef>* pStartingConfiguration,
Reference<TLSOptions> tlsOptions, int extraDB, std::string whitelistBinPaths) {
int extraDB, std::string whitelistBinPaths) {
CSimpleIni ini;
ini.SetUnicode();
ini.LoadFile(joinPath(baseFolder, "restartInfo.ini").c_str());
@ -710,7 +644,7 @@ ACTOR Future<Void> restartSimulatedSystem(vector<Future<Void>>* systemActors, st
// SOMEDAY: parse backup agent from test file
systemActors->push_back(reportErrors(
simulatedMachine(conn, ipAddrs, usingSSL, tlsOptions, localities, processClass, baseFolder, true,
simulatedMachine(conn, ipAddrs, usingSSL, localities, processClass, baseFolder, true,
i == useSeedForMachine, enableExtraDB ? AgentAddition : AgentNone,
usingSSL && (listenersPerProcess == 1 || processClass == ProcessClass::TesterClass), whitelistBinPaths),
processClass == ProcessClass::TesterClass ? "SimulatedTesterMachine" : "SimulatedMachine"));
@ -1087,8 +1021,7 @@ void SimulationConfig::generateNormalConfig(int minimumReplication, int minimumR
void setupSimulatedSystem(vector<Future<Void>>* systemActors, std::string baseFolder, int* pTesterCount,
Optional<ClusterConnectionString>* pConnString, Standalone<StringRef>* pStartingConfiguration,
int extraDB, int minimumReplication, int minimumRegions, Reference<TLSOptions> tlsOptions,
std::string whitelistBinPaths, bool configureLocked) {
int extraDB, int minimumReplication, int minimumRegions, std::string whitelistBinPaths, bool configureLocked) {
// SOMEDAY: this does not test multi-interface configurations
SimulationConfig simconfig(extraDB, minimumReplication, minimumRegions);
StatusObject startingConfigJSON = simconfig.db.toJSON(true);
@ -1162,7 +1095,7 @@ void setupSimulatedSystem(vector<Future<Void>>* systemActors, std::string baseFo
bool assignClasses = machineCount - dataCenters > 4 && deterministicRandom()->random01() < 0.5;
// Use SSL 5% of the time
bool sslEnabled = deterministicRandom()->random01() < 0.10 && tlsOptions->enabled();
bool sslEnabled = deterministicRandom()->random01() < 0.10;
bool sslOnly = sslEnabled && deterministicRandom()->coinflip();
g_simulator.listenersPerProcess = sslEnabled && !sslOnly ? 2 : 1;
TEST( sslEnabled ); // SSL enabled
@ -1221,9 +1154,9 @@ void setupSimulatedSystem(vector<Future<Void>>* systemActors, std::string baseFo
.detail("Address", coordinatorAddresses[i])
.detail("Coordinators", describe(coordinatorAddresses));
g_simulator.protectedAddresses.insert(
NetworkAddress(coordinatorAddresses[i].ip, coordinatorAddresses[i].port, true, false));
NetworkAddress(coordinatorAddresses[i].ip, coordinatorAddresses[i].port, true, coordinatorAddresses[i].isTLS()));
if(coordinatorAddresses[i].port==2) {
g_simulator.protectedAddresses.insert(NetworkAddress(coordinatorAddresses[i].ip, 1, true, false));
g_simulator.protectedAddresses.insert(NetworkAddress(coordinatorAddresses[i].ip, 1, true, true));
}
}
deterministicRandom()->randomShuffle(coordinatorAddresses);
@ -1292,7 +1225,7 @@ void setupSimulatedSystem(vector<Future<Void>>* systemActors, std::string baseFo
// check the sslEnablementMap using only one ip(
LocalityData localities(Optional<Standalone<StringRef>>(), zoneId, machineId, dcUID);
localities.set(LiteralStringRef("data_hall"), dcUID);
systemActors->push_back(reportErrors(simulatedMachine(conn, ips, sslEnabled, tlsOptions,
systemActors->push_back(reportErrors(simulatedMachine(conn, ips, sslEnabled,
localities, processClass, baseFolder, false, machine == useSeedForMachine, requiresExtraDBMachines ? AgentOnly : AgentAddition, sslOnly, whitelistBinPaths ), "SimulatedMachine"));
if (requiresExtraDBMachines) {
@ -1305,7 +1238,7 @@ void setupSimulatedSystem(vector<Future<Void>>* systemActors, std::string baseFo
LocalityData localities(Optional<Standalone<StringRef>>(), newZoneId, newMachineId, dcUID);
localities.set(LiteralStringRef("data_hall"), dcUID);
systemActors->push_back(reportErrors(simulatedMachine(*g_simulator.extraDB, extraIps, sslEnabled, tlsOptions,
systemActors->push_back(reportErrors(simulatedMachine(*g_simulator.extraDB, extraIps, sslEnabled,
localities,
processClass, baseFolder, false, machine == useSeedForMachine, AgentNone, sslOnly, whitelistBinPaths ), "SimulatedMachine"));
}
@ -1333,9 +1266,9 @@ void setupSimulatedSystem(vector<Future<Void>>* systemActors, std::string baseFo
Standalone<StringRef> newZoneId = Standalone<StringRef>(deterministicRandom()->randomUniqueID().toString());
LocalityData localities(Optional<Standalone<StringRef>>(), newZoneId, newZoneId, Optional<Standalone<StringRef>>());
systemActors->push_back( reportErrors( simulatedMachine(
conn, ips, sslEnabled, tlsOptions,
conn, ips, sslEnabled && sslOnly,
localities, ProcessClass(ProcessClass::TesterClass, ProcessClass::CommandLineSource),
baseFolder, false, i == useSeedForMachine, AgentNone, sslEnabled, whitelistBinPaths ),
baseFolder, false, i == useSeedForMachine, AgentNone, sslEnabled && sslOnly, whitelistBinPaths ),
"SimulatedTesterMachine") );
}
*pStartingConfiguration = startingConfigString;
@ -1396,7 +1329,7 @@ void checkTestConf(const char* testFile, int& extraDB, int& minimumReplication,
ifs.close();
}
ACTOR void setupAndRun(std::string dataFolder, const char *testFile, bool rebooting, bool restoring, std::string whitelistBinPaths, Reference<TLSOptions> tlsOptions) {
ACTOR void setupAndRun(std::string dataFolder, const char *testFile, bool rebooting, bool restoring, std::string whitelistBinPaths) {
state vector<Future<Void>> systemActors;
state Optional<ClusterConnectionString> connFile;
state Standalone<StringRef> startingConfiguration;
@ -1409,7 +1342,7 @@ ACTOR void setupAndRun(std::string dataFolder, const char *testFile, bool reboot
// TODO (IPv6) Use IPv6?
wait(g_simulator.onProcess(
g_simulator.newProcess("TestSystem", IPAddress(0x01010101), 1, 1,
g_simulator.newProcess("TestSystem", IPAddress(0x01010101), 1, false, 1,
LocalityData(Optional<Standalone<StringRef>>(),
Standalone<StringRef>(deterministicRandom()->randomUniqueID().toString()),
Standalone<StringRef>(deterministicRandom()->randomUniqueID().toString()),
@ -1418,16 +1351,12 @@ ACTOR void setupAndRun(std::string dataFolder, const char *testFile, bool reboot
TaskPriority::DefaultYield));
Sim2FileSystem::newFileSystem();
FlowTransport::createInstance(true, 1);
if (tlsOptions->enabled()) {
simInitTLS(tlsOptions);
}
TEST(true); // Simulation start
try {
//systemActors.push_back( startSystemMonitor(dataFolder) );
if (rebooting) {
wait( timeoutError( restartSimulatedSystem( &systemActors, dataFolder, &testerCount, &connFile, &startingConfiguration, tlsOptions, extraDB, whitelistBinPaths), 100.0 ) );
wait( timeoutError( restartSimulatedSystem( &systemActors, dataFolder, &testerCount, &connFile, &startingConfiguration, extraDB, whitelistBinPaths), 100.0 ) );
// FIXME: snapshot restore does not support multi-region restore, hence restore it as single region always
if (restoring) {
startingConfiguration = LiteralStringRef("usable_regions=1");
@ -1436,7 +1365,7 @@ ACTOR void setupAndRun(std::string dataFolder, const char *testFile, bool reboot
else {
g_expect_full_pointermap = 1;
setupSimulatedSystem(&systemActors, dataFolder, &testerCount, &connFile, &startingConfiguration, extraDB,
minimumReplication, minimumRegions, tlsOptions, whitelistBinPaths, configureLocked);
minimumReplication, minimumRegions, whitelistBinPaths, configureLocked);
wait( delay(1.0) ); // FIXME: WHY!!! //wait for machines to boot
}
std::string clusterFileDir = joinPath( dataFolder, deterministicRandom()->randomUniqueID().toString() );

4
fdbserver/SimulatedCluster.h
View File

@ -18,12 +18,10 @@
* limitations under the License.
*/
#include "fdbrpc/TLSConnection.h"
#ifndef FDBSERVER_SIMULATEDCLUSTER_H
#define FDBSERVER_SIMULATEDCLUSTER_H
#pragma once
void setupAndRun(std::string const& dataFolder, const char* const& testFile, bool const& rebooting, bool const& restoring, std::string const& whitelistBinPath, Reference<TLSOptions> const& useSSL);
void setupAndRun(std::string const& dataFolder, const char* const& testFile, bool const& rebooting, bool const& restoring, std::string const& whitelistBinPath);
#endif

55
fdbserver/fdbserver.actor.cpp
View File

@ -52,12 +52,12 @@
#include "fdbserver/workloads/workloads.actor.h"
#include <time.h>
#include "fdbserver/Status.h"
#include "fdbrpc/TLSConnection.h"
#include "fdbrpc/Net2FileSystem.h"
#include "fdbrpc/Platform.h"
#include "fdbrpc/AsyncFileCached.actor.h"
#include "fdbserver/CoroFlow.h"
#include "flow/SignalSafeUnwind.h"
#include "flow/TLSPolicy.h"
#if defined(CMAKE_BUILD) || !defined(WIN32)
#include "versions.h"
#endif
@ -961,8 +961,8 @@ int main(int argc, char* argv[]) {
int minTesterCount = 1;
bool testOnServers = false;
Reference<TLSOptions> tlsOptions = Reference<TLSOptions>( new TLSOptions );
std::string tlsCertPath, tlsKeyPath, tlsCAPath, tlsPassword;
Reference<TLSPolicy> tlsPolicy = Reference<TLSPolicy>(new TLSPolicy(TLSPolicy::Is::SERVER));
TLSParams tlsParams;
std::vector<std::string> tlsVerifyPeers;
double fileIoTimeout = 0.0;
bool fileIoWarnOnly = false;
@ -1331,22 +1331,22 @@ int main(int argc, char* argv[]) {
whitelistBinPaths = args.OptionArg();
break;
#ifndef TLS_DISABLED
case TLSOptions::OPT_TLS_PLUGIN:
case TLSParams::OPT_TLS_PLUGIN:
args.OptionArg();
break;
case TLSOptions::OPT_TLS_CERTIFICATES:
tlsCertPath = args.OptionArg();
case TLSParams::OPT_TLS_CERTIFICATES:
tlsParams.tlsCertPath = args.OptionArg();
break;
case TLSOptions::OPT_TLS_PASSWORD:
tlsPassword = args.OptionArg();
case TLSParams::OPT_TLS_PASSWORD:
tlsParams.tlsPassword = args.OptionArg();
break;
case TLSOptions::OPT_TLS_CA_FILE:
tlsCAPath = args.OptionArg();
case TLSParams::OPT_TLS_CA_FILE:
tlsParams.tlsCAPath = args.OptionArg();
break;
case TLSOptions::OPT_TLS_KEY:
tlsKeyPath = args.OptionArg();
case TLSParams::OPT_TLS_KEY:
tlsParams.tlsKeyPath = args.OptionArg();
break;
case TLSOptions::OPT_TLS_VERIFY_PEERS:
case TLSParams::OPT_TLS_VERIFY_PEERS:
tlsVerifyPeers.push_back(args.OptionArg());
break;
#endif
@ -1551,7 +1551,12 @@ int main(int argc, char* argv[]) {
startNewSimulator();
openTraceFile(NetworkAddress(), rollsize, maxLogsSize, logFolder, "trace", logGroup);
} else {
g_network = newNet2(useThreadPool, true);
#ifndef TLS_DISABLED
if ( tlsVerifyPeers.size() ) {
tlsPolicy->set_verify_peers( tlsVerifyPeers );
}
#endif
g_network = newNet2(useThreadPool, true, tlsPolicy, tlsParams);
FlowTransport::createInstance(false, 1);
const bool expectsPublicAddress = (role == FDBD || role == NetworkTestServer || role == Restore);
@ -1565,22 +1570,7 @@ int main(int argc, char* argv[]) {
openTraceFile(publicAddresses.address, rollsize, maxLogsSize, logFolder, "trace", logGroup);
#ifndef TLS_DISABLED
if ( tlsCertPath.size() )
tlsOptions->set_cert_file( tlsCertPath );
if (tlsCAPath.size())
tlsOptions->set_ca_file(tlsCAPath);
if (tlsKeyPath.size()) {
if (tlsPassword.size())
tlsOptions->set_key_password(tlsPassword);
tlsOptions->set_key_file(tlsKeyPath);
}
if ( tlsVerifyPeers.size() )
tlsOptions->set_verify_peers( tlsVerifyPeers );
tlsOptions->register_network();
#endif
if (expectsPublicAddress) {
for (int ii = 0; ii < (publicAddresses.secondaryAddress.present() ? 2 : 1); ++ii) {
const NetworkAddress& publicAddress = ii==0 ? publicAddresses.address : publicAddresses.secondaryAddress.get();
@ -1789,7 +1779,7 @@ int main(int argc, char* argv[]) {
}
}
}
setupAndRun( dataFolder, testFile, restarting, (isRestoring >= 1), whitelistBinPaths, tlsOptions);
setupAndRun( dataFolder, testFile, restarting, (isRestoring >= 1), whitelistBinPaths);
g_simulator.run();
} else if (role == FDBD) {
ASSERT( connectionFile );
@ -1973,6 +1963,11 @@ int main(int argc, char* argv[]) {
TraceEvent(SevError, "MainError").error(e);
//printf("\n%d tests passed; %d tests failed\n", passCount, failCount);
flushAndExit(FDB_EXIT_MAIN_ERROR);
} catch (boost::system::system_error& e) {
fprintf(stderr, "boost::system::system_error: %s (%d)", e.what(), e.code().value());
TraceEvent(SevError, "MainError").error(unknown_error()).detail("RootException", e.what());
//printf("\n%d tests passed; %d tests failed\n", passCount, failCount);
flushAndExit(FDB_EXIT_MAIN_EXCEPTION);
} catch (std::exception& e) {
fprintf(stderr, "std::exception: %s\n", e.what());
TraceEvent(SevError, "MainError").error(unknown_error()).detail("RootException", e.what());

12
fdbserver/workloads/ConsistencyCheck.actor.cpp
View File

@ -1137,18 +1137,19 @@ struct ConsistencyCheckWorkload : TestWorkload
std::set<Optional<Key>> missingStorage;
for( int i = 0; i < workers.size(); i++ ) {
if( !configuration.isExcludedServer(workers[i].interf.address()) &&
NetworkAddress addr = workers[i].interf.tLog.getEndpoint().addresses.getTLSAddress();
if( !configuration.isExcludedServer(addr) &&
( workers[i].processClass == ProcessClass::StorageClass || workers[i].processClass == ProcessClass::UnsetClass ) ) {
bool found = false;
for( int j = 0; j < storageServers.size(); j++ ) {
if( storageServers[j].address() == workers[i].interf.address() ) {
if( storageServers[j].getVersion.getEndpoint().addresses.getTLSAddress() == addr ) {
found = true;
break;
}
}
if( !found ) {
TraceEvent("ConsistencyCheck_NoStorage")
.detail("Address", workers[i].interf.address())
.detail("Address", addr)
.detail("ProcessClassEqualToStorageClass",
(int)(workers[i].processClass == ProcessClass::StorageClass));
missingStorage.insert(workers[i].interf.locality.dcId());
@ -1221,12 +1222,13 @@ struct ConsistencyCheckWorkload : TestWorkload
std::set<NetworkAddress> workerAddresses;
for (const auto& it : workers) {
ISimulator::ProcessInfo* info = g_simulator.getProcessByAddress(it.interf.address());
NetworkAddress addr = it.interf.tLog.getEndpoint().addresses.getTLSAddress();
ISimulator::ProcessInfo* info = g_simulator.getProcessByAddress(addr);
if(!info || info->failed) {
TraceEvent("ConsistencyCheck_FailedWorkerInList").detail("Addr", it.interf.address());
return false;
}
workerAddresses.insert( NetworkAddress(it.interf.address().ip, it.interf.address().port, true, false) );
workerAddresses.insert( NetworkAddress(addr.ip, addr.port, true, addr.isTLS()) );
}
vector<ISimulator::ProcessInfo*> all = g_simulator.getAllProcesses();

12
flow/CMakeLists.txt
View File

@ -58,6 +58,8 @@ set(FLOW_SRCS
ThreadSafeQueue.h
Trace.cpp
Trace.h
TLSPolicy.h
TLSPolicy.cpp
UnitTest.cpp
UnitTest.h
XmlTraceLogFormatter.h
@ -90,14 +92,14 @@ elseif(WIN32)
target_link_libraries(flow PUBLIC psapi.lib)
endif()
target_link_libraries(flow PRIVATE ${FLOW_LIBS})
target_link_libraries(flow PUBLIC boost_target Threads::Threads ${CMAKE_DL_LIBS})
if(USE_VALGRIND)
target_link_libraries(flow PUBLIC Valgrind)
endif()
if(NOT WITH_TLS)
target_compile_definitions(flow PUBLIC TLS_DISABLED)
else()
target_link_libraries(flow PUBLIC FDBLibTLS)
target_link_libraries(flow PUBLIC OpenSSL::SSL)
endif()
target_link_libraries(flow PUBLIC boost_target Threads::Threads ${CMAKE_DL_LIBS})
if(USE_VALGRIND)
target_link_libraries(flow PUBLIC Valgrind)
endif()
if(APPLE)

4
flow/Knobs.cpp
View File

@ -67,7 +67,6 @@ FlowKnobs::FlowKnobs(bool randomize, bool isSimulated) {
init( MAX_RECONNECTION_TIME, 0.5 );
init( RECONNECTION_TIME_GROWTH_RATE, 1.2 );
init( RECONNECTION_RESET_TIME, 5.0 );
init( CONNECTION_ACCEPT_DELAY, 0.5 );
init( USE_OBJECT_SERIALIZER, 1 );
init( TOO_MANY_CONNECTIONS_CLOSED_RESET_DELAY, 5.0 );
init( TOO_MANY_CONNECTIONS_CLOSED_TIMEOUT, 20.0 );
@ -117,6 +116,7 @@ FlowKnobs::FlowKnobs(bool randomize, bool isSimulated) {
init( SLOW_LOOP_CUTOFF, 15.0 / 1000.0 );
init( SLOW_LOOP_SAMPLING_RATE, 0.1 );
init( TSC_YIELD_TIME, 1000000 );
init( CERT_FILE_MAX_SIZE, 5 * 1024 * 1024 );
//Network
init( PACKET_LIMIT, 100LL<<20 );
@ -125,6 +125,8 @@ FlowKnobs::FlowKnobs(bool randomize, bool isSimulated) {
init( MAX_PACKET_SEND_BYTES, 256 * 1024 );
init( MIN_PACKET_BUFFER_BYTES, 4 * 1024 );
init( MIN_PACKET_BUFFER_FREE_BYTES, 256 );
init( UNRESTRICTED_HANDSHAKE_LIMIT, 15 );
init( BOUNDED_HANDSHAKE_LIMIT, 400 );
//Sim2
init( MIN_OPEN_TIME, 0.0002 );

4
flow/Knobs.h
View File

@ -87,7 +87,6 @@ public:
double MAX_RECONNECTION_TIME;
double RECONNECTION_TIME_GROWTH_RATE;
double RECONNECTION_RESET_TIME;
double CONNECTION_ACCEPT_DELAY;
int USE_OBJECT_SERIALIZER;
int TLS_CERT_REFRESH_DELAY_SECONDS;
@ -138,6 +137,7 @@ public:
double SLOW_LOOP_SAMPLING_RATE;
int64_t TSC_YIELD_TIME;
int64_t REACTOR_FLAGS;
int CERT_FILE_MAX_SIZE;
//Network
int64_t PACKET_LIMIT;
@ -146,6 +146,8 @@ public:
int MAX_PACKET_SEND_BYTES;
int MIN_PACKET_BUFFER_BYTES;
int MIN_PACKET_BUFFER_FREE_BYTES;
int UNRESTRICTED_HANDSHAKE_LIMIT;
int BOUNDED_HANDSHAKE_LIMIT;
//Sim2
//FIMXE: more parameters could be factored out

428
flow/Net2.actor.cpp
View File

@ -37,6 +37,7 @@
#include "flow/AsioReactor.h"
#include "flow/Profiler.h"
#include "flow/ProtocolVersion.h"
#include "flow/TLSPolicy.h"
#ifdef WIN32
#include <mmsystem.h>
@ -49,7 +50,6 @@ intptr_t g_stackYieldLimit = 0;
using namespace boost::asio::ip;
#if defined(__linux__)
#include <execinfo.h>
@ -111,7 +111,7 @@ thread_local INetwork* thread_network = 0;
class Net2 sealed : public INetwork, public INetworkConnections {
public:
Net2(bool useThreadPool, bool useMetrics);
Net2(bool useThreadPool, bool useMetrics, Reference<TLSPolicy> policy, const TLSParams& tlsParams);
void run();
void initMetrics();
@ -155,6 +155,15 @@ public:
//private:
ASIOReactor reactor;
#ifndef TLS_DISABLED
boost::asio::ssl::context sslContext;
#endif
std::string tlsPassword;
std::string get_password() const {
return tlsPassword;
}
INetworkConnections *network; // initially this, but can be changed
int64_t tsc_begin, tsc_end;
@ -245,7 +254,11 @@ public:
try {
if (error) {
// Log the error...
TraceEvent(SevWarn, errContext, errID).suppressFor(1.0).detail("ErrorCode", error.value()).detail("Message", error.message());
TraceEvent(SevWarn, errContext, errID).suppressFor(1.0).detail("ErrorCode", error.value()).detail("Message", error.message())
#ifndef TLS_DISABLED
.detail("WhichMeans", TLSPolicy::ErrorString(error))
#endif
;
p.sendError( connection_failed() );
} else
p.send( Void() );
@ -298,6 +311,10 @@ public:
init();
}
virtual Future<Void> acceptHandshake() { return Void(); }
virtual Future<Void> connectHandshake() { return Void(); }
// returns when write() can write at least one byte
virtual Future<Void> onWritable() {
++g_net2->countWriteProbes;
@ -472,6 +489,342 @@ private:
}
};
#ifndef TLS_DISABLED
typedef boost::asio::ssl::stream<boost::asio::ip::tcp::socket&> ssl_socket;
class SSLConnection : public IConnection, ReferenceCounted<SSLConnection> {
public:
virtual void addref() { ReferenceCounted<SSLConnection>::addref(); }
virtual void delref() { ReferenceCounted<SSLConnection>::delref(); }
virtual void close() {
closeSocket();
}
explicit SSLConnection( boost::asio::io_service& io_service, boost::asio::ssl::context& context )
: id(nondeterministicRandom()->randomUniqueID()), socket(io_service), ssl_sock(socket, context)
{
}
// This is not part of the IConnection interface, because it is wrapped by INetwork::connect()
ACTOR static Future<Reference<IConnection>> connect( boost::asio::io_service* ios, boost::asio::ssl::context* context, NetworkAddress addr ) {
std::pair<IPAddress,uint16_t> peerIP = std::make_pair(addr.ip, addr.port);
auto iter(g_network->networkInfo.serverTLSConnectionThrottler.find(peerIP));
if(iter != g_network->networkInfo.serverTLSConnectionThrottler.end()) {
if (now() < iter->second.second) {
if(iter->second.first >= FLOW_KNOBS->TLS_CLIENT_CONNECTION_THROTTLE_ATTEMPTS) {
TraceEvent("TLSOutgoingConnectionThrottlingWarning").suppressFor(1.0).detail("PeerIP", addr);
wait(delay(FLOW_KNOBS->CONNECTION_MONITOR_TIMEOUT));
throw connection_failed();
}
} else {
g_network->networkInfo.serverTLSConnectionThrottler.erase(peerIP);
}
}
state Reference<SSLConnection> self( new SSLConnection(*ios, *context) );
self->peer_address = addr;
try {
auto to = tcpEndpoint(self->peer_address);
BindPromise p("N2_ConnectError", self->id);
Future<Void> onConnected = p.getFuture();
self->socket.async_connect( to, std::move(p) );
wait( onConnected );
self->init();
return self;
} catch (Error& e) {
// Either the connection failed, or was cancelled by the caller
self->closeSocket();
throw;
}
}
// This is not part of the IConnection interface, because it is wrapped by IListener::accept()
void accept(NetworkAddress peerAddr) {
this->peer_address = peerAddr;
init();
}
ACTOR static void doAcceptHandshake( Reference<SSLConnection> self, Promise<Void> connected) {
try {
state std::pair<IPAddress,uint16_t> peerIP = std::make_pair(self->getPeerAddress().ip, static_cast<uint16_t>(0));
auto iter(g_network->networkInfo.serverTLSConnectionThrottler.find(peerIP));
if(iter != g_network->networkInfo.serverTLSConnectionThrottler.end()) {
if (now() < iter->second.second) {
if(iter->second.first >= FLOW_KNOBS->TLS_SERVER_CONNECTION_THROTTLE_ATTEMPTS) {
TraceEvent("TLSIncomingConnectionThrottlingWarning").suppressFor(1.0).detail("PeerIP", peerIP.first.toString());
wait(delay(FLOW_KNOBS->CONNECTION_MONITOR_TIMEOUT));
self->closeSocket();
connected.sendError(connection_failed());
return;
}
} else {
g_network->networkInfo.serverTLSConnectionThrottler.erase(peerIP);
}
}
int64_t permitNumber = wait(g_network->networkInfo.handshakeLock->take());
state BoundedFlowLock::Releaser releaser(g_network->networkInfo.handshakeLock, permitNumber);
BindPromise p("N2_AcceptHandshakeError", UID());
auto onHandshook = p.getFuture();
self->getSSLSocket().async_handshake( boost::asio::ssl::stream_base::server, std::move(p) );
wait( onHandshook );
wait(delay(0, TaskPriority::Handshake));
connected.send(Void());
} catch (...) {
auto iter(g_network->networkInfo.serverTLSConnectionThrottler.find(peerIP));
if(iter != g_network->networkInfo.serverTLSConnectionThrottler.end()) {
iter->second.first++;
} else {
g_network->networkInfo.serverTLSConnectionThrottler[peerIP] = std::make_pair(0,now() + FLOW_KNOBS->TLS_SERVER_CONNECTION_THROTTLE_TIMEOUT);
}
self->closeSocket();
connected.sendError(connection_failed());
}
}
ACTOR static Future<Void> acceptHandshakeWrapper( Reference<SSLConnection> self ) {
Promise<Void> connected;
doAcceptHandshake(self, connected);
try {
wait(connected.getFuture());
return Void();
} catch (Error& e) {
// Either the connection failed, or was cancelled by the caller
self->closeSocket();
throw;
}
}
virtual Future<Void> acceptHandshake() {
return acceptHandshakeWrapper( Reference<SSLConnection>::addRef(this) );
}
ACTOR static void doConnectHandshake( Reference<SSLConnection> self, Promise<Void> connected) {
try {
int64_t permitNumber = wait(g_network->networkInfo.handshakeLock->take());
state BoundedFlowLock::Releaser releaser(g_network->networkInfo.handshakeLock, permitNumber);
BindPromise p("N2_ConnectHandshakeError", self->id);
Future<Void> onHandshook = p.getFuture();
self->ssl_sock.async_handshake( boost::asio::ssl::stream_base::client, std::move(p) );
wait( onHandshook );
wait(delay(0, TaskPriority::Handshake));
connected.send(Void());
} catch (...) {
std::pair<IPAddress,uint16_t> peerIP = std::make_pair(self->peer_address.ip, self->peer_address.port);
auto iter(g_network->networkInfo.serverTLSConnectionThrottler.find(peerIP));
if(iter != g_network->networkInfo.serverTLSConnectionThrottler.end()) {
iter->second.first++;
} else {
g_network->networkInfo.serverTLSConnectionThrottler[peerIP] = std::make_pair(0,now() + FLOW_KNOBS->TLS_CLIENT_CONNECTION_THROTTLE_TIMEOUT);
}
self->closeSocket();
connected.sendError(connection_failed());
}
}
ACTOR static Future<Void> connectHandshakeWrapper( Reference<SSLConnection> self ) {
Promise<Void> connected;
doConnectHandshake(self, connected);
try {
wait(connected.getFuture());
return Void();
} catch (Error& e) {
// Either the connection failed, or was cancelled by the caller
self->closeSocket();
throw;
}
}
virtual Future<Void> connectHandshake() {
return connectHandshakeWrapper( Reference<SSLConnection>::addRef(this) );
}
// returns when write() can write at least one byte
virtual Future<Void> onWritable() {
++g_net2->countWriteProbes;
BindPromise p("N2_WriteProbeError", id);
auto f = p.getFuture();
socket.async_write_some( boost::asio::null_buffers(), std::move(p) );
return f;
}
// returns when read() can read at least one byte
virtual Future<Void> onReadable() {
++g_net2->countReadProbes;
BindPromise p("N2_ReadProbeError", id);
auto f = p.getFuture();
socket.async_read_some( boost::asio::null_buffers(), std::move(p) );
return f;
}
// Reads as many bytes as possible from the read buffer into [begin,end) and returns the number of bytes read (might be 0)
virtual int read( uint8_t* begin, uint8_t* end ) {
boost::system::error_code err;
++g_net2->countReads;
size_t toRead = end-begin;
size_t size = ssl_sock.read_some( boost::asio::mutable_buffers_1(begin, toRead), err );
g_net2->bytesReceived += size;
//TraceEvent("ConnRead", this->id).detail("Bytes", size);
if (err) {
if (err == boost::asio::error::would_block) {
++g_net2->countWouldBlock;
return 0;
}
onReadError(err);
throw connection_failed();
}
ASSERT( size ); // If the socket is closed, we expect an 'eof' error, not a zero return value
return size;
}
// Writes as many bytes as possible from the given SendBuffer chain into the write buffer and returns the number of bytes written (might be 0)
virtual int write( SendBuffer const* data, int limit ) {
boost::system::error_code err;
++g_net2->countWrites;
size_t sent = ssl_sock.write_some( boost::iterator_range<SendBufferIterator>(SendBufferIterator(data, limit), SendBufferIterator()), err );
if (err) {
// Since there was an error, sent's value can't be used to infer that the buffer has data and the limit is positive so check explicitly.
ASSERT(limit > 0);
bool notEmpty = false;
for(auto p = data; p; p = p->next)
if(p->bytes_written - p->bytes_sent > 0) {
notEmpty = true;
break;
}
ASSERT(notEmpty);
if (err == boost::asio::error::would_block) {
++g_net2->countWouldBlock;
return 0;
}
onWriteError(err);
throw connection_failed();
}
ASSERT( sent ); // Make sure data was sent, and also this check will fail if the buffer chain was empty or the limit was not > 0.
return sent;
}
virtual NetworkAddress getPeerAddress() { return peer_address; }
virtual UID getDebugID() { return id; }
tcp::socket& getSocket() { return socket; }
ssl_socket& getSSLSocket() { return ssl_sock; }
private:
UID id;
tcp::socket socket;
ssl_socket ssl_sock;
NetworkAddress peer_address;
struct SendBufferIterator {
typedef boost::asio::const_buffer value_type;
typedef std::forward_iterator_tag iterator_category;
typedef size_t difference_type;
typedef boost::asio::const_buffer* pointer;
typedef boost::asio::const_buffer& reference;
SendBuffer const* p;
int limit;
SendBufferIterator(SendBuffer const* p=0, int limit = std::numeric_limits<int>::max()) : p(p), limit(limit) {
ASSERT(limit > 0);
}
bool operator == (SendBufferIterator const& r) const { return p == r.p; }
bool operator != (SendBufferIterator const& r) const { return p != r.p; }
void operator++() {
limit -= p->bytes_written - p->bytes_sent;
if(limit > 0)
p = p->next;
else
p = NULL;
}
boost::asio::const_buffer operator*() const {
return boost::asio::const_buffer( p->data + p->bytes_sent, std::min(limit, p->bytes_written - p->bytes_sent) );
}
};
void init() {
// Socket settings that have to be set after connect or accept succeeds
socket.non_blocking(true);
socket.set_option(boost::asio::ip::tcp::no_delay(true));
platform::setCloseOnExec(socket.native_handle());
}
void closeSocket() {
boost::system::error_code cancelError;
socket.cancel(cancelError);
boost::system::error_code closeError;
socket.close(closeError);
boost::system::error_code shutdownError;
ssl_sock.shutdown(shutdownError);
}
void onReadError( const boost::system::error_code& error ) {
TraceEvent(SevWarn, "N2_ReadError", id).suppressFor(1.0).detail("Message", error.value());
closeSocket();
}
void onWriteError( const boost::system::error_code& error ) {
TraceEvent(SevWarn, "N2_WriteError", id).suppressFor(1.0).detail("Message", error.value());
closeSocket();
}
};
class SSLListener : public IListener, ReferenceCounted<SSLListener> {
NetworkAddress listenAddress;
tcp::acceptor acceptor;
boost::asio::ssl::context* context;
public:
SSLListener( boost::asio::io_service& io_service, boost::asio::ssl::context* context, NetworkAddress listenAddress )
: listenAddress(listenAddress), acceptor( io_service, tcpEndpoint( listenAddress ) ), context(context)
{
platform::setCloseOnExec(acceptor.native_handle());
}
virtual void addref() { ReferenceCounted<SSLListener>::addref(); }
virtual void delref() { ReferenceCounted<SSLListener>::delref(); }
// Returns one incoming connection when it is available
virtual Future<Reference<IConnection>> accept() {
return doAccept( this );
}
virtual NetworkAddress getListenAddress() { return listenAddress; }
private:
ACTOR static Future<Reference<IConnection>> doAccept( SSLListener* self ) {
state Reference<SSLConnection> conn( new SSLConnection( self->acceptor.get_io_service(), *self->context) );
state tcp::acceptor::endpoint_type peer_endpoint;
try {
BindPromise p("N2_AcceptError", UID());
auto f = p.getFuture();
self->acceptor.async_accept( conn->getSocket(), peer_endpoint, std::move(p) );
wait( f );
auto peer_address = peer_endpoint.address().is_v6() ? IPAddress(peer_endpoint.address().to_v6().to_bytes()) : IPAddress(peer_endpoint.address().to_v4().to_ulong());
conn->accept(NetworkAddress(peer_address, peer_endpoint.port(), false, true));
return conn;
} catch (...) {
conn->close();
throw;
}
}
};
#endif
struct PromiseTask : public Task, public FastAllocated<PromiseTask> {
Promise<Void> promise;
PromiseTask() {}
@ -483,7 +836,15 @@ struct PromiseTask : public Task, public FastAllocated<PromiseTask> {
}
};
Net2::Net2(bool useThreadPool, bool useMetrics)
// 5MB for loading files into memory
#ifndef TLS_DISABLED
bool insecurely_always_accept(bool _1, boost::asio::ssl::verify_context& _2) {
return true;
}
#endif
Net2::Net2(bool useThreadPool, bool useMetrics, Reference<TLSPolicy> policy, const TLSParams& tlsParams)
: useThreadPool(useThreadPool),
network(this),
reactor(this),
@ -492,10 +853,49 @@ Net2::Net2(bool useThreadPool, bool useMetrics)
// Until run() is called, yield() will always yield
tsc_begin(0), tsc_end(0), taskBegin(0), currentTaskID(TaskPriority::DefaultYield),
lastMinTaskID(TaskPriority::Zero),
numYields(0)
numYields(0),
tlsPassword(tlsParams.tlsPassword)
#ifndef TLS_DISABLED
,sslContext(boost::asio::ssl::context(boost::asio::ssl::context::tlsv12))
#endif
{
TraceEvent("Net2Starting");
#ifndef TLS_DISABLED
sslContext.set_options(boost::asio::ssl::context::default_workarounds);
sslContext.set_verify_mode(boost::asio::ssl::context::verify_peer | boost::asio::ssl::verify_fail_if_no_peer_cert);
if (policy) {
sslContext.set_verify_callback([policy](bool preverified, boost::asio::ssl::verify_context& ctx) {
return policy->verify_peer(preverified, ctx.native_handle());
});
} else {
sslContext.set_verify_callback(boost::bind(&insecurely_always_accept, _1, _2));
}
sslContext.set_password_callback(std::bind(&Net2::get_password, this));
if (tlsParams.tlsCertPath.size() ) {
sslContext.use_certificate_chain_file(tlsParams.tlsCertPath);
}
if (tlsParams.tlsCertBytes.size() ) {
sslContext.use_certificate(boost::asio::buffer(tlsParams.tlsCertBytes.data(), tlsParams.tlsCertBytes.size()), boost::asio::ssl::context::pem);
}
if (tlsParams.tlsCAPath.size()) {
std::string cert = readFileBytes(tlsParams.tlsCAPath, FLOW_KNOBS->CERT_FILE_MAX_SIZE);
sslContext.add_certificate_authority(boost::asio::buffer(cert.data(), cert.size()));
}
if (tlsParams.tlsCABytes.size()) {
sslContext.add_certificate_authority(boost::asio::buffer(tlsParams.tlsCABytes.data(), tlsParams.tlsCABytes.size()));
}
if (tlsParams.tlsKeyPath.size()) {
sslContext.use_private_key_file(tlsParams.tlsKeyPath, boost::asio::ssl::context::pem);
}
if (tlsParams.tlsKeyBytes.size()) {
sslContext.use_private_key(boost::asio::buffer(tlsParams.tlsKeyBytes.data(), tlsParams.tlsKeyBytes.size()), boost::asio::ssl::context::pem);
}
#endif
// Set the global members
if(useMetrics) {
setGlobal(INetwork::enTDMetrics, (flowGlobalType) &tdmetrics);
@ -871,8 +1271,13 @@ THREAD_HANDLE Net2::startThread( THREAD_FUNC_RETURN (*func) (void*), void *arg )
return ::startThread(func, arg);
}
Future< Reference<IConnection> > Net2::connect( NetworkAddress toAddr, std::string host ) {
#ifndef TLS_DISABLED
if ( toAddr.isTLS() ) {
return SSLConnection::connect(&this->reactor.ios, &this->sslContext, toAddr);
}
#endif
return Connection::connect(&this->reactor.ios, toAddr);
}
@ -946,6 +1351,11 @@ bool Net2::isAddressOnThisHost( NetworkAddress const& addr ) {
Reference<IListener> Net2::listen( NetworkAddress localAddr ) {
try {
#ifndef TLS_DISABLED
if ( localAddr.isTLS() ) {
return Reference<IListener>(new SSLListener( reactor.ios, &this->sslContext, localAddr ));
}
#endif
return Reference<IListener>( new Listener( reactor.ios, localAddr ) );
} catch (boost::system::system_error const& e) {
Error x;
@ -1040,13 +1450,13 @@ void ASIOReactor::wake() {
} // namespace net2
INetwork* newNet2(bool useThreadPool, bool useMetrics) {
INetwork* newNet2(bool useThreadPool, bool useMetrics, Reference<TLSPolicy> policy, const TLSParams& tlsParams) {
try {
N2::g_net2 = new N2::Net2(useThreadPool, useMetrics);
N2::g_net2 = new N2::Net2(useThreadPool, useMetrics, policy, tlsParams);
}
catch(boost::system::system_error e) {
TraceEvent("Net2InitError").detail("Message", e.what());
throw unknown_error();
throw;
}
catch(std::exception const& e) {
TraceEvent("Net2InitError").detail("Message", e.what());

535
flow/TLSPolicy.cpp Normal file
View File

@ -0,0 +1,535 @@
/*
* TLSPolicy.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2020 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include "flow/TLSPolicy.h"
TLSPolicy::~TLSPolicy() {}
#ifndef TLS_DISABLED
#include <algorithm>
#include <cstring>
#include <exception>
#include <map>
#include <set>
#include <openssl/objects.h>
#include <openssl/bio.h>
#include <openssl/err.h>
#include <openssl/pem.h>
#include <openssl/x509.h>
#include <openssl/x509v3.h>
#include <openssl/x509_vfy.h>
#include <stdint.h>
#include <string>
#include <sstream>
#include <utility>
#include "flow/FastRef.h"
#include "flow/Trace.h"
std::string TLSPolicy::ErrorString(boost::system::error_code e) {
char* str = ERR_error_string(e.value(), NULL);
return std::string(str);
}
// To force typeinfo to only be emitted once.
std::string TLSPolicy::toString() const {
std::stringstream ss;
ss << "TLSPolicy{ Rules=[";
for (const auto &r : rules) {
ss << " " << r.toString() << ",";
}
ss << " ] }";
return ss.str();
}
std::string TLSPolicy::Rule::toString() const {
std::stringstream ss;
ss << "Rule{ verify_cert=" << verify_cert
<< ", verify_time=" << verify_time;
ss << ", Subject=[";
for (const auto& s : subject_criteria) {
ss << " { NID=" << s.first << ", Criteria=" << s.second.criteria << "},";
}
ss << " ], Issuer=[";
for (const auto& s : issuer_criteria) {
ss << " { NID=" << s.first << ", Criteria=" << s.second.criteria << "},";
}
ss << " ], Root=[";
for (const auto& s : root_criteria) {
ss << " { NID=" << s.first << ", Criteria=" << s.second.criteria << "},";
}
ss << " ] }";
return ss.str();
}
static int hexValue(char c) {
static char const digits[] = "0123456789ABCDEF";
if (c >= 'a' && c <= 'f')
c -= ('a' - 'A');
int value = std::find(digits, digits + 16, c) - digits;
if (value >= 16) {
throw std::runtime_error("hexValue");
}
return value;
}
// Does not handle "raw" form (e.g. #28C4D1), only escaped text
static std::string de4514(std::string const& input, int start, int& out_end) {
std::string output;
if(input[start] == '#' || input[start] == ' ') {
out_end = start;
return output;
}
int space_count = 0;
for(int p = start; p < input.size();) {
switch(input[p]) {
case '\\': // Handle escaped sequence
// Backslash escaping nothing!
if(p == input.size() - 1) {
out_end = p;
goto FIN;
}
switch(input[p+1]) {
case ' ':
case '"':
case '#':
case '+':
case ',':
case ';':
case '<':
case '=':
case '>':
case '|':
case '\\':
output += input[p+1];
p += 2;
space_count = 0;
continue;
default:
// Backslash escaping pair of hex digits requires two characters
if(p == input.size() - 2) {
out_end = p;
goto FIN;
}
try {
output += hexValue(input[p+1]) * 16 + hexValue(input[p+2]);
p += 3;
space_count = 0;
continue;
} catch( ... ) {
out_end = p;
goto FIN;
}
}
case '"':
case '+':
case ',':
case ';':
case '<':
case '>':
case 0:
// All of these must have been escaped
out_end = p;
goto FIN;
default:
// Character is what it is
output += input[p];
if(input[p] == ' ')
space_count++;
else
space_count = 0;
p++;
}
}
out_end = input.size();
FIN:
out_end -= space_count;
output.resize(output.size() - space_count);
return output;
}
static std::pair<std::string, std::string> splitPair(std::string const& input, char c) {
int p = input.find_first_of(c);
if(p == input.npos) {
throw std::runtime_error("splitPair");
}
return std::make_pair(input.substr(0, p), input.substr(p+1, input.size()));
}
static NID abbrevToNID(std::string const& sn) {
NID nid = NID_undef;
if (sn == "C" || sn == "CN" || sn == "L" || sn == "ST" || sn == "O" || sn == "OU" || sn == "UID" || sn == "DC" || sn == "subjectAltName")
nid = OBJ_sn2nid(sn.c_str());
if (nid == NID_undef)
throw std::runtime_error("abbrevToNID");
return nid;
}
static X509Location locationForNID(NID nid) {
const char* name = OBJ_nid2ln(nid);
if (name == NULL) {
throw std::runtime_error("locationForNID");
}
if (strncmp(name, "X509v3", 6) == 0) {
return X509Location::EXTENSION;
} else {
// It probably isn't true that all other NIDs live in the NAME, but it is for now...
return X509Location::NAME;
}
}
bool TLSPolicy::set_verify_peers(std::vector<std::string> verify_peers) {
for (int i = 0; i < verify_peers.size(); i++) {
try {
std::string& verifyString = verify_peers[i];
int start = 0;
while(start < verifyString.size()) {
int split = verifyString.find('|', start);
if(split == std::string::npos) {
break;
}
if(split == start || verifyString[split-1] != '\\') {
rules.emplace_back(verifyString.substr(start,split-start));
start = split+1;
}
}
rules.emplace_back(verifyString.substr(start));
} catch ( const std::runtime_error& e ) {
rules.clear();
std::string& verifyString = verify_peers[i];
TraceEvent(SevError, "FDBLibTLSVerifyPeersParseError").detail("Config", verifyString);
return false;
}
}
return true;
}
TLSPolicy::Rule::Rule(std::string input) {
int s = 0;
while (s < input.size()) {
int eq = input.find('=', s);
if (eq == input.npos)
throw std::runtime_error("parse_verify");
MatchType mt = MatchType::EXACT;
if (input[eq-1] == '>') mt = MatchType::PREFIX;
if (input[eq-1] == '<') mt = MatchType::SUFFIX;
std::string term = input.substr(s, eq - s - (mt == MatchType::EXACT ? 0 : 1));
if (term.find("Check.") == 0) {
if (eq + 2 > input.size())
throw std::runtime_error("parse_verify");
if (eq + 2 != input.size() && input[eq + 2] != ',')
throw std::runtime_error("parse_verify");
if (mt != MatchType::EXACT)
throw std::runtime_error("parse_verify: cannot prefix match Check");
bool* flag;
if (term == "Check.Valid")
flag = &verify_cert;
else if (term == "Check.Unexpired")
flag = &verify_time;
else
throw std::runtime_error("parse_verify");
if (input[eq + 1] == '0')
*flag = false;
else if (input[eq + 1] == '1')
*flag = true;
else
throw std::runtime_error("parse_verify");
s = eq + 3;
} else {
std::map< int, Criteria >* criteria = &subject_criteria;
if (term.find('.') != term.npos) {
auto scoped = splitPair(term, '.');
if (scoped.first == "S" || scoped.first == "Subject")
criteria = &subject_criteria;
else if (scoped.first == "I" || scoped.first == "Issuer")
criteria = &issuer_criteria;
else if (scoped.first == "R" || scoped.first == "Root")
criteria = &root_criteria;
else
throw std::runtime_error("parse_verify");
term = scoped.second;
}
int remain;
auto unesc = de4514(input, eq + 1, remain);
if (remain == eq + 1)
throw std::runtime_error("parse_verify");
NID termNID = abbrevToNID(term);
const X509Location loc = locationForNID(termNID);
criteria->insert(std::make_pair(termNID, Criteria(unesc, mt, loc)));
if (remain != input.size() && input[remain] != ',')
throw std::runtime_error("parse_verify");
s = remain + 1;
}
}
}
bool match_criteria_entry(const std::string& criteria, ASN1_STRING* entry, MatchType mt) {
bool rc = false;
ASN1_STRING* asn_criteria = NULL;
unsigned char* criteria_utf8 = NULL;
int criteria_utf8_len = 0;
unsigned char* entry_utf8 = NULL;
int entry_utf8_len = 0;
if ((asn_criteria = ASN1_IA5STRING_new()) == NULL)
goto err;
if (ASN1_STRING_set(asn_criteria, criteria.c_str(), criteria.size()) != 1)
goto err;
if ((criteria_utf8_len = ASN1_STRING_to_UTF8(&criteria_utf8, asn_criteria)) < 1)
goto err;
if ((entry_utf8_len = ASN1_STRING_to_UTF8(&entry_utf8, entry)) < 1)
goto err;
if (mt == MatchType::EXACT) {
if (criteria_utf8_len == entry_utf8_len &&
memcmp(criteria_utf8, entry_utf8, criteria_utf8_len) == 0)
rc = true;
} else if (mt == MatchType::PREFIX) {
if (criteria_utf8_len <= entry_utf8_len &&
memcmp(criteria_utf8, entry_utf8, criteria_utf8_len) == 0)
rc = true;
} else if (mt == MatchType::SUFFIX) {
if (criteria_utf8_len <= entry_utf8_len &&
memcmp(criteria_utf8, entry_utf8 + (entry_utf8_len - criteria_utf8_len), criteria_utf8_len) == 0)
rc = true;
}
err:
ASN1_STRING_free(asn_criteria);
free(criteria_utf8);
free(entry_utf8);
return rc;
}
bool match_name_criteria(X509_NAME *name, NID nid, const std::string& criteria, MatchType mt) {
X509_NAME_ENTRY *name_entry;
int idx;
// If name does not exist, or has multiple of this RDN, refuse to proceed.
if ((idx = X509_NAME_get_index_by_NID(name, nid, -1)) < 0)
return false;
if (X509_NAME_get_index_by_NID(name, nid, idx) != -1)
return false;
if ((name_entry = X509_NAME_get_entry(name, idx)) == NULL)
return false;
return match_criteria_entry(criteria, X509_NAME_ENTRY_get_data(name_entry), mt);
}
bool match_extension_criteria(X509 *cert, NID nid, const std::string& value, MatchType mt) {
if (nid != NID_subject_alt_name && nid != NID_issuer_alt_name) {
// I have no idea how other extensions work.
return false;
}
auto pos = value.find(':');
if (pos == value.npos) {
return false;
}
std::string value_gen = value.substr(0, pos);
std::string value_val = value.substr(pos+1, value.npos);
STACK_OF(GENERAL_NAME)* sans = reinterpret_cast<STACK_OF(GENERAL_NAME)*>(X509_get_ext_d2i(cert, nid, NULL, NULL));
if (sans == NULL) {
return false;
}
int num_sans = sk_GENERAL_NAME_num( sans );
bool rc = false;
for( int i = 0; i < num_sans && !rc; ++i ) {
GENERAL_NAME* altname = sk_GENERAL_NAME_value( sans, i );
std::string matchable;
switch (altname->type) {
case GEN_OTHERNAME:
break;
case GEN_EMAIL:
if (value_gen == "EMAIL" &&
match_criteria_entry( value_val, altname->d.rfc822Name, mt)) {
rc = true;
break;
}
case GEN_DNS:
if (value_gen == "DNS" &&
match_criteria_entry( value_val, altname->d.dNSName, mt )) {
rc = true;
break;
}
case GEN_X400:
case GEN_DIRNAME:
case GEN_EDIPARTY:
break;
case GEN_URI:
if (value_gen == "URI" &&
match_criteria_entry( value_val, altname->d.uniformResourceIdentifier, mt )) {
rc = true;
break;
}
case GEN_IPADD:
if (value_gen == "IP" &&
match_criteria_entry( value_val, altname->d.iPAddress, mt )) {
rc = true;
break;
}
case GEN_RID:
break;
}
}
sk_GENERAL_NAME_pop_free(sans, GENERAL_NAME_free);
return rc;
}
bool match_criteria(X509* cert, X509_NAME* subject, NID nid, const std::string& criteria, MatchType mt, X509Location loc) {
switch(loc) {
case X509Location::NAME: {
return match_name_criteria(subject, nid, criteria, mt);
}
case X509Location::EXTENSION: {
return match_extension_criteria(cert, nid, criteria, mt);
}
}
// Should never be reachable.
return false;
}
std::tuple<bool,std::string> check_verify(const TLSPolicy::Rule* verify, X509_STORE_CTX* store_ctx, bool is_client) {
X509_NAME *subject, *issuer;
bool rc = false;
X509* cert = NULL;
// if returning false, give a reason string
std::string reason = "";
// Check subject criteria.
cert = sk_X509_value(X509_STORE_CTX_get0_chain(store_ctx), 0);
if ((subject = X509_get_subject_name(cert)) == NULL) {
reason = "Cert subject error";
goto err;
}
for (auto &pair: verify->subject_criteria) {
if (!match_criteria(cert, subject, pair.first, pair.second.criteria, pair.second.match_type, pair.second.location)) {
reason = "Cert subject match failure";
goto err;
}
}
// Check issuer criteria.
if ((issuer = X509_get_issuer_name(cert)) == NULL) {
reason = "Cert issuer error";
goto err;
}
for (auto &pair: verify->issuer_criteria) {
if (!match_criteria(cert, issuer, pair.first, pair.second.criteria, pair.second.match_type, pair.second.location)) {
reason = "Cert issuer match failure";
goto err;
}
}
// Check root criteria - this is the subject of the final certificate in the stack.
cert = sk_X509_value(X509_STORE_CTX_get0_chain(store_ctx), sk_X509_num(X509_STORE_CTX_get0_chain(store_ctx)) - 1);
if ((subject = X509_get_subject_name(cert)) == NULL) {
reason = "Root subject error";
goto err;
}
for (auto &pair: verify->root_criteria) {
if (!match_criteria(cert, subject, pair.first, pair.second.criteria, pair.second.match_type, pair.second.location)) {
reason = "Root subject match failure";
goto err;
}
}
// If we got this far, everything checked out...
rc = true;
err:
return std::make_tuple(rc, reason);
}
bool TLSPolicy::verify_peer(bool preverified, X509_STORE_CTX* store_ctx) {
bool rc = false;
std::set<std::string> verify_failure_reasons;
bool verify_success;
std::string verify_failure_reason;
// If certificate verification is disabled, there's nothing more to do.
if (std::any_of(rules.begin(), rules.end(), [](const Rule& r){ return !r.verify_cert; })) {
return true;
}
if(!preverified) {
TraceEvent("TLSPolicyFailure").suppressFor(1.0).detail("Reason", "preverification failed").detail("VerifyError", X509_verify_cert_error_string(X509_STORE_CTX_get_error(store_ctx)));
return false;
}
if(!rules.size()) {
return true;
}
// Any matching rule is sufficient.
for (auto &verify_rule: rules) {
std::tie(verify_success, verify_failure_reason) = check_verify(&verify_rule, store_ctx, is_client);
if (verify_success) {
rc = true;
break;
} else {
if (verify_failure_reason.length() > 0)
verify_failure_reasons.insert(verify_failure_reason);
}
}
if (!rc) {
// log the various failure reasons
for (std::string reason : verify_failure_reasons) {
TraceEvent("TLSPolicyFailure").suppressFor(1.0).detail("Reason", reason);
}
}
return rc;
}
#endif

145
flow/TLSPolicy.h Normal file
View File

@ -0,0 +1,145 @@
/*
* TLSPolicy.h
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2020 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#ifndef _FLOW_TLSPOLICY_H_
#define _FLOW_TLSPOLICY_H_
#pragma once
#include <map>
#include <string>
#include <vector>
#include <boost/system/system_error.hpp>
#include "flow/FastRef.h"
#ifndef TLS_DISABLED
#include <openssl/x509.h>
typedef int NID;
enum class MatchType {
EXACT,
PREFIX,
SUFFIX,
};
enum class X509Location {
// This NID is located within a X509_NAME
NAME,
// This NID is an X509 extension, and should be parsed accordingly
EXTENSION,
};
struct Criteria {
Criteria( const std::string& s )
: criteria(s), match_type(MatchType::EXACT), location(X509Location::NAME) {}
Criteria( const std::string& s, MatchType mt )
: criteria(s), match_type(mt), location(X509Location::NAME) {}
Criteria( const std::string& s, X509Location loc)
: criteria(s), match_type(MatchType::EXACT), location(loc) {}
Criteria( const std::string& s, MatchType mt, X509Location loc)
: criteria(s), match_type(mt), location(loc) {}
std::string criteria;
MatchType match_type;
X509Location location;
bool operator==(const Criteria& c) const {
return criteria == c.criteria && match_type == c.match_type && location == c.location;
}
};
#endif
struct TLSParams {
enum { OPT_TLS = 100000, OPT_TLS_PLUGIN, OPT_TLS_CERTIFICATES, OPT_TLS_KEY, OPT_TLS_VERIFY_PEERS, OPT_TLS_CA_FILE, OPT_TLS_PASSWORD };
std::string tlsCertPath, tlsKeyPath, tlsCAPath, tlsPassword;
std::string tlsCertBytes, tlsKeyBytes, tlsCABytes;
};
class TLSPolicy : ReferenceCounted<TLSPolicy> {
public:
enum class Is {
CLIENT,
SERVER
};
TLSPolicy(Is client) : is_client(client == Is::CLIENT) {}
virtual ~TLSPolicy();
virtual void addref() { ReferenceCounted<TLSPolicy>::addref(); }
virtual void delref() { ReferenceCounted<TLSPolicy>::delref(); }
#ifndef TLS_DISABLED
static std::string ErrorString(boost::system::error_code e);
bool set_verify_peers(std::vector<std::string> verify_peers);
bool verify_peer(bool preverified, X509_STORE_CTX* store_ctx);
std::string toString() const;
struct Rule {
explicit Rule(std::string input);
std::string toString() const;
std::map< NID, Criteria > subject_criteria;
std::map< NID, Criteria > issuer_criteria;
std::map< NID, Criteria > root_criteria;
bool verify_cert = true;
bool verify_time = true;
};
std::vector<Rule> rules;
#endif
bool is_client;
};
#define TLS_PLUGIN_FLAG "--tls_plugin"
#define TLS_CERTIFICATE_FILE_FLAG "--tls_certificate_file"
#define TLS_KEY_FILE_FLAG "--tls_key_file"
#define TLS_VERIFY_PEERS_FLAG "--tls_verify_peers"
#define TLS_CA_FILE_FLAG "--tls_ca_file"
#define TLS_PASSWORD_FLAG "--tls_password"
#define TLS_OPTION_FLAGS \
{ TLSParams::OPT_TLS_PLUGIN, TLS_PLUGIN_FLAG, SO_REQ_SEP }, \
{ TLSParams::OPT_TLS_CERTIFICATES, TLS_CERTIFICATE_FILE_FLAG, SO_REQ_SEP }, \
{ TLSParams::OPT_TLS_KEY, TLS_KEY_FILE_FLAG, SO_REQ_SEP }, \
{ TLSParams::OPT_TLS_VERIFY_PEERS, TLS_VERIFY_PEERS_FLAG, SO_REQ_SEP }, \
{ TLSParams::OPT_TLS_PASSWORD, TLS_PASSWORD_FLAG, SO_REQ_SEP }, \
{ TLSParams::OPT_TLS_CA_FILE, TLS_CA_FILE_FLAG, SO_REQ_SEP },
#define TLS_HELP \
" " TLS_CERTIFICATE_FILE_FLAG " CERTFILE\n" \
" The path of a file containing the TLS certificate and CA\n" \
" chain.\n" \
" " TLS_CA_FILE_FLAG " CERTAUTHFILE\n" \
" The path of a file containing the CA certificates chain.\n" \
" " TLS_KEY_FILE_FLAG " KEYFILE\n" \
" The path of a file containing the private key corresponding\n" \
" to the TLS certificate.\n" \
" " TLS_PASSWORD_FLAG " PASSCODE\n" \
" The passphrase of encrypted private key\n" \
" " TLS_VERIFY_PEERS_FLAG " CONSTRAINTS\n" \
" The constraints by which to validate TLS peers. The contents\n" \
" and format of CONSTRAINTS are plugin-specific.\n"
#endif

2
flow/flow.vcxproj
View File

@ -49,6 +49,7 @@
<ClCompile Include="version.cpp" />
<ClCompile Include="SignalSafeUnwind.cpp" />
<ClCompile Include="serialize.cpp" />
<ClCompile Include="TLSPolicy.cpp" />
</ItemGroup>
<ItemGroup>
<ClInclude Include="CompressedInt.h" />
@ -92,6 +93,7 @@
<ClInclude Include="Platform.h" />
<ClInclude Include="ThreadSafeQueue.h" />
<ClInclude Include="Trace.h" />
<ClInclude Include="TLSPolicy.h" />
<ClInclude Include="SignalSafeUnwind.h" />
<ClInclude Include="UnitTest.h" />
<ActorCompiler Include="ThreadHelper.actor.h">

105
flow/genericactors.actor.h
View File

@ -32,6 +32,7 @@
#include "flow/flow.h"
#include "flow/Knobs.h"
#include "flow/Util.h"
#include "flow/IndexedSet.h"
#include "flow/actorcompiler.h" // This must be the last #include.
#pragma warning( disable: 4355 ) // 'this' : used in base member initializer list
@ -1340,6 +1341,110 @@ private:
}
};
struct NotifiedInt {
NotifiedInt( int64_t val = 0 ) : val(val) {}
Future<Void> whenAtLeast( int64_t limit ) {
if (val >= limit)
return Void();
Promise<Void> p;
waiting.push( std::make_pair(limit,p) );
return p.getFuture();
}
int64_t get() const { return val; }
void set( int64_t v ) {
ASSERT( v >= val );
if (v != val) {
val = v;
std::vector<Promise<Void>> toSend;
while ( waiting.size() && v >= waiting.top().first ) {
Promise<Void> p = std::move(waiting.top().second);
waiting.pop();
toSend.push_back(p);
}
for(auto& p : toSend) {
p.send(Void());
}
}
}
void operator=( int64_t v ) {
set( v );
}
NotifiedInt(NotifiedInt&& r) BOOST_NOEXCEPT : waiting(std::move(r.waiting)), val(r.val) {}
void operator=(NotifiedInt&& r) BOOST_NOEXCEPT { waiting = std::move(r.waiting); val = r.val; }
private:
typedef std::pair<int64_t,Promise<Void>> Item;
struct ItemCompare {
bool operator()(const Item& a, const Item& b) { return a.first > b.first; }
};
std::priority_queue<Item, std::vector<Item>, ItemCompare> waiting;
int64_t val;
};
struct BoundedFlowLock : NonCopyable, public ReferenceCounted<BoundedFlowLock> {
// BoundedFlowLock is different from a FlowLock in that it has a bound on how many locks can be taken from the oldest outstanding lock.
// For instance, with a FlowLock that has two permits, if one permit is taken but never released, the other permit can be reused an unlimited
// amount of times, but with a BoundedFlowLock, it can only be reused a fixed number of times.
struct Releaser : NonCopyable {
BoundedFlowLock* lock;
int64_t permitNumber;
Releaser() : lock(nullptr), permitNumber(0) {}
Releaser( BoundedFlowLock* lock, int64_t permitNumber ) : lock(lock), permitNumber(permitNumber) {}
Releaser(Releaser&& r) BOOST_NOEXCEPT : lock(r.lock), permitNumber(r.permitNumber) { r.permitNumber = 0; }
void operator=(Releaser&& r) { if (permitNumber) lock->release(permitNumber); lock = r.lock; permitNumber = r.permitNumber; r.permitNumber = 0; }
void release() {
if (permitNumber) {
lock->release(permitNumber);
}
permitNumber = 0;
}
~Releaser() { if (permitNumber) lock->release(permitNumber); }
};
BoundedFlowLock() : unrestrictedPermits(1), boundedPermits(0), nextPermitNumber(0), minOutstanding(0) {}
explicit BoundedFlowLock(int64_t unrestrictedPermits, int64_t boundedPermits) : unrestrictedPermits(unrestrictedPermits), boundedPermits(boundedPermits), nextPermitNumber(0), minOutstanding(0) {}
Future<int64_t> take() {
return takeActor(this);
}
void release( int64_t permitNumber ) {
outstanding.erase(permitNumber);
updateMinOutstanding();
}
private:
IndexedSet<int64_t, int64_t> outstanding;
NotifiedInt minOutstanding;
int64_t nextPermitNumber;
const int64_t unrestrictedPermits;
const int64_t boundedPermits;
void updateMinOutstanding() {
auto it = outstanding.index(unrestrictedPermits-1);
if(it == outstanding.end()) {
minOutstanding.set(nextPermitNumber);
} else {
minOutstanding.set(*it);
}
}
ACTOR static Future<int64_t> takeActor(BoundedFlowLock* lock) {
state int64_t permitNumber = ++lock->nextPermitNumber;
lock->outstanding.insert(permitNumber, 1);
lock->updateMinOutstanding();
wait( lock->minOutstanding.whenAtLeast(std::max<int64_t>(0, permitNumber - lock->boundedPermits)) );
return permitNumber;
}
};
ACTOR template <class T>
Future<Void> yieldPromiseStream( FutureStream<T> input, PromiseStream<T> output, TaskPriority taskID = TaskPriority::DefaultYield ) {
loop {

2
flow/network.cpp
View File

@ -200,3 +200,5 @@ TEST_CASE("/flow/network/ipaddress") {
return Void();
}
NetworkInfo::NetworkInfo() : handshakeLock( new BoundedFlowLock(FLOW_KNOBS->UNRESTRICTED_HANDSHAKE_LIMIT, FLOW_KNOBS->BOUNDED_HANDSHAKE_LIMIT) ) {}

24
flow/network.h
View File

@ -27,8 +27,12 @@
#include <stdint.h>
#include <variant>
#include "boost/asio.hpp"
#ifndef TLS_DISABLED
#include "boost/asio/ssl.hpp"
#endif
#include "flow/serialize.h"
#include "flow/IRandom.h"
#include "flow/TLSPolicy.h"
enum class TaskPriority {
Max = 1000000,
@ -40,6 +44,8 @@ enum class TaskPriority {
DiskIOComplete = 9150,
LoadBalancedEndpoint = 9000,
ReadSocket = 9000,
AcceptSocket = 8950,
Handshake = 8900,
CoordinationReply = 8810,
Coordination = 8800,
FailureMonitor = 8700,
@ -281,6 +287,13 @@ struct NetworkAddressList {
return secondaryAddress < r.secondaryAddress;
}
NetworkAddress getTLSAddress() const {
if(!secondaryAddress.present() || address.isTLS()) {
return address;
}
return secondaryAddress.get();
}
std::string toString() const {
if(!secondaryAddress.present()) {
return address.toString();
@ -319,6 +332,8 @@ struct NetworkMetrics {
NetworkMetrics() {}
};
struct BoundedFlowLock;
struct NetworkInfo {
NetworkMetrics metrics;
double oldestAlternativesFailure = 0;
@ -326,8 +341,9 @@ struct NetworkInfo {
double lastAlternativesFailureSkipDelay = 0;
std::map<std::pair<IPAddress, uint16_t>, std::pair<int,double>> serverTLSConnectionThrottler;
BoundedFlowLock* handshakeLock;
NetworkInfo() {}
NetworkInfo();
};
class IEventFD : public ReferenceCounted<IEventFD> {
@ -346,6 +362,10 @@ public:
// Closes the underlying connection eventually if it is not already closed.
virtual void close() = 0;
virtual Future<Void> acceptHandshake() = 0;
virtual Future<Void> connectHandshake() = 0;
// returns when write() can write at least one byte (or may throw an error if the connection dies)
virtual Future<Void> onWritable() = 0;
@ -389,7 +409,7 @@ typedef NetworkAddressList (*NetworkAddressesFuncPtr)();
class INetwork;
extern INetwork* g_network;
extern INetwork* newNet2(bool useThreadPool = false, bool useMetrics = false);
extern INetwork* newNet2(bool useThreadPool = false, bool useMetrics = false, Reference<TLSPolicy> policy = Reference<TLSPolicy>(), const TLSParams& tlsParams = TLSParams());
class INetwork {
public: