solution: tls for gprc upstream
This commit is contained in:
@@ -49,7 +49,7 @@ dependencies {
|
|||||||
compile "io.grpc:grpc-protobuf:${grpcVersion}"
|
compile "io.grpc:grpc-protobuf:${grpcVersion}"
|
||||||
compile "io.grpc:grpc-stub:${grpcVersion}"
|
compile "io.grpc:grpc-stub:${grpcVersion}"
|
||||||
compile "io.grpc:grpc-netty:${grpcVersion}"
|
compile "io.grpc:grpc-netty:${grpcVersion}"
|
||||||
compile "io.netty:netty-tcnative-boringssl-static:2.0.22.Final"
|
compile "io.netty:netty-tcnative-boringssl-static:2.0.25.Final"
|
||||||
compile "io.netty:netty-all:4.1.36.Final"
|
compile "io.netty:netty-all:4.1.36.Final"
|
||||||
|
|
||||||
compile "org.jetbrains.kotlin:kotlin-stdlib-jdk8"
|
compile "org.jetbrains.kotlin:kotlin-stdlib-jdk8"
|
||||||
|
|||||||
@@ -122,7 +122,8 @@ open class ConfiguredUpstreams(
|
|||||||
endpoint.host!!,
|
endpoint.host!!,
|
||||||
endpoint.port ?: 443,
|
endpoint.port ?: 443,
|
||||||
objectMapper,
|
objectMapper,
|
||||||
options
|
options,
|
||||||
|
up.auth
|
||||||
)
|
)
|
||||||
log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}")
|
log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}")
|
||||||
ds.start()
|
ds.start()
|
||||||
|
|||||||
@@ -6,7 +6,12 @@ import io.emeraldpay.api.proto.ReactorBlockchainGrpc
|
|||||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||||
import io.emeraldpay.grpc.Chain
|
import io.emeraldpay.grpc.Chain
|
||||||
import io.grpc.ManagedChannelBuilder
|
import io.grpc.ManagedChannelBuilder
|
||||||
|
import io.grpc.netty.NettyChannelBuilder
|
||||||
|
import io.netty.handler.ssl.*
|
||||||
|
import org.apache.commons.lang3.StringUtils
|
||||||
|
import org.slf4j.LoggerFactory
|
||||||
import reactor.core.publisher.Mono
|
import reactor.core.publisher.Mono
|
||||||
|
import java.io.File
|
||||||
import java.util.*
|
import java.util.*
|
||||||
import java.util.concurrent.locks.ReentrantLock
|
import java.util.concurrent.locks.ReentrantLock
|
||||||
import kotlin.concurrent.withLock
|
import kotlin.concurrent.withLock
|
||||||
@@ -15,17 +20,26 @@ class GrpcUpstreams(
|
|||||||
private val host: String,
|
private val host: String,
|
||||||
private val port: Int,
|
private val port: Int,
|
||||||
private val objectMapper: ObjectMapper,
|
private val objectMapper: ObjectMapper,
|
||||||
private val options: UpstreamsConfig.Options
|
private val options: UpstreamsConfig.Options,
|
||||||
|
private val auth: UpstreamsConfig.TlsAuth? = null
|
||||||
) {
|
) {
|
||||||
|
private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java)
|
||||||
|
|
||||||
private var client: ReactorBlockchainGrpc.ReactorBlockchainStub? = null
|
private var client: ReactorBlockchainGrpc.ReactorBlockchainStub? = null
|
||||||
private var known = HashMap<Chain, GrpcUpstream>()
|
private var known = HashMap<Chain, GrpcUpstream>()
|
||||||
private val lock = ReentrantLock()
|
private val lock = ReentrantLock()
|
||||||
|
|
||||||
fun start(): Mono<List<Chain>> {
|
fun start(): Mono<List<Chain>> {
|
||||||
val channel = ManagedChannelBuilder.forAddress(host, port)
|
val channel: ManagedChannelBuilder<*> = if (auth != null && StringUtils.isNotEmpty(auth.ca)) {
|
||||||
.enableRetry()
|
NettyChannelBuilder.forAddress(host, port)
|
||||||
channel.usePlaintext()
|
.useTransportSecurity()
|
||||||
|
.sslContext(withTls(auth))
|
||||||
|
} else {
|
||||||
|
log.warn("Using insecure connection for $host:$port")
|
||||||
|
ManagedChannelBuilder.forAddress(host, port)
|
||||||
|
.usePlaintext()
|
||||||
|
}
|
||||||
|
|
||||||
val client = ReactorBlockchainGrpc.newReactorStub(channel.build())
|
val client = ReactorBlockchainGrpc.newReactorStub(channel.build())
|
||||||
this.client = client
|
this.client = client
|
||||||
val loaded = client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build())
|
val loaded = client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build())
|
||||||
@@ -52,6 +66,24 @@ class GrpcUpstreams(
|
|||||||
return loaded
|
return loaded
|
||||||
}
|
}
|
||||||
|
|
||||||
|
internal fun withTls(auth: UpstreamsConfig.TlsAuth): SslContext {
|
||||||
|
val sslContext = SslContextBuilder.forClient()
|
||||||
|
.clientAuth(ClientAuth.REQUIRE)
|
||||||
|
sslContext.trustManager(File(auth.ca!!).inputStream())
|
||||||
|
if (StringUtils.isNotEmpty(auth.key) && StringUtils.isNoneEmpty(auth.certificate)) {
|
||||||
|
sslContext.keyManager(File(auth.certificate!!).inputStream(), File(auth.key!!).inputStream())
|
||||||
|
} else {
|
||||||
|
log.warn("Connect to remote using only CA certificate")
|
||||||
|
}
|
||||||
|
val alpn = ApplicationProtocolConfig(
|
||||||
|
ApplicationProtocolConfig.Protocol.ALPN,
|
||||||
|
ApplicationProtocolConfig.SelectorFailureBehavior.NO_ADVERTISE,
|
||||||
|
ApplicationProtocolConfig.SelectedListenerFailureBehavior.ACCEPT,
|
||||||
|
"grpc-exp", "h2")
|
||||||
|
sslContext.applicationProtocolConfig(alpn)
|
||||||
|
return sslContext.build()
|
||||||
|
}
|
||||||
|
|
||||||
fun getOrCreate(chain: Chain): GrpcUpstream {
|
fun getOrCreate(chain: Chain): GrpcUpstream {
|
||||||
lock.withLock {
|
lock.withLock {
|
||||||
val current = known[chain]
|
val current = known[chain]
|
||||||
|
|||||||
Reference in New Issue
Block a user