Add client auth (#294)
This commit is contained in:
@@ -84,7 +84,7 @@ open class GrpcServer(
|
||||
}
|
||||
|
||||
serverBuilder.intercept(grpcServerBraveInterceptor)
|
||||
if (mainConfig.authorization.enabled) {
|
||||
if (mainConfig.authorization.enabled && mainConfig.authorization.hasServerConfig()) {
|
||||
serverBuilder.intercept(authInterceptor)
|
||||
log.info("Token authorization is turned on")
|
||||
}
|
||||
|
||||
@@ -22,7 +22,8 @@ class AuthService(
|
||||
}
|
||||
|
||||
val keys = rsaKeyReader.getKeyPair(
|
||||
authorizationConfig.providerPrivateKeyPath, authorizationConfig.externalPublicKeyPath
|
||||
authorizationConfig.serverConfig.providerPrivateKeyPath,
|
||||
authorizationConfig.serverConfig.externalPublicKeyPath
|
||||
)
|
||||
val decodedJwt = JWT.decode(token)
|
||||
|
||||
|
||||
@@ -39,6 +39,10 @@ class AuthConfig {
|
||||
var key: String? = null
|
||||
) : ClientAuth()
|
||||
|
||||
class ClientTokenAuth(
|
||||
var publicKeyPath: String? = null
|
||||
)
|
||||
|
||||
/**
|
||||
* Example config:
|
||||
* ```
|
||||
|
||||
@@ -49,6 +49,14 @@ class AuthConfigReader : YamlConfigReader<AuthConfig>() {
|
||||
}
|
||||
}
|
||||
|
||||
fun readTokenAuth(node: MappingNode?): AuthConfig.ClientTokenAuth? {
|
||||
return getMapping(node, "token-auth")?.let {
|
||||
val auth = AuthConfig.ClientTokenAuth()
|
||||
auth.publicKeyPath = getValueAsString(it, "public-key")
|
||||
auth
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Example config:
|
||||
* ```
|
||||
|
||||
@@ -3,12 +3,37 @@ package io.emeraldpay.dshackle.config
|
||||
data class AuthorizationConfig(
|
||||
val enabled: Boolean,
|
||||
val publicKeyOwner: String,
|
||||
val providerPrivateKeyPath: String,
|
||||
val externalPublicKeyPath: String
|
||||
val serverConfig: ServerConfig,
|
||||
val clientConfig: ClientConfig
|
||||
) {
|
||||
|
||||
fun hasServerConfig() = serverConfig != ServerConfig.default()
|
||||
|
||||
companion object {
|
||||
@JvmStatic
|
||||
fun default() = AuthorizationConfig(false, "", "", "")
|
||||
fun default() = AuthorizationConfig(
|
||||
false, "",
|
||||
ServerConfig.default(),
|
||||
ClientConfig.default()
|
||||
)
|
||||
}
|
||||
|
||||
data class ServerConfig(
|
||||
val providerPrivateKeyPath: String,
|
||||
val externalPublicKeyPath: String
|
||||
) {
|
||||
companion object {
|
||||
@JvmStatic
|
||||
fun default() = ServerConfig("", "")
|
||||
}
|
||||
}
|
||||
|
||||
data class ClientConfig(
|
||||
val privateKeyPath: String,
|
||||
) {
|
||||
companion object {
|
||||
@JvmStatic
|
||||
fun default() = ClientConfig("")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -27,20 +27,38 @@ class AuthorizationConfigReader : YamlConfigReader<AuthorizationConfig>() {
|
||||
val publicKeyOwner = getValueAsString(auth, "publicKeyOwner")
|
||||
?: throw IllegalStateException("Public key owner in not specified")
|
||||
|
||||
val keyPair = getMapping(auth, "keys") ?: throw IllegalStateException("Auth keys is not specified")
|
||||
val privateKey = getValueAsString(keyPair, "provider-private-key")
|
||||
?: throw IllegalStateException("Private key in not specified")
|
||||
val publicKey = getValueAsString(keyPair, "external-public-key")
|
||||
?: throw IllegalStateException("External key in not specified")
|
||||
val authServer = getMapping(auth, "server")
|
||||
?.run {
|
||||
val keyPair = getMapping(this, "keys")
|
||||
?: throw IllegalStateException("Auth keys is not specified")
|
||||
val privateKey = getValueAsString(keyPair, "provider-private-key")
|
||||
?: throw IllegalStateException("Private key in not specified")
|
||||
val publicKey = getValueAsString(keyPair, "external-public-key")
|
||||
?: throw IllegalStateException("External key in not specified")
|
||||
|
||||
if (fileNotExists(privateKey)) {
|
||||
throw IllegalStateException("There is no such file: $privateKey")
|
||||
}
|
||||
if (fileNotExists(publicKey)) {
|
||||
throw IllegalStateException("There is no such file: $publicKey")
|
||||
if (fileNotExists(privateKey)) {
|
||||
throw IllegalStateException("There is no such file: $privateKey")
|
||||
}
|
||||
if (fileNotExists(publicKey)) {
|
||||
throw IllegalStateException("There is no such file: $publicKey")
|
||||
}
|
||||
AuthorizationConfig.ServerConfig(privateKey, publicKey)
|
||||
}
|
||||
|
||||
val authClient = getMapping(auth, "client")
|
||||
?.run {
|
||||
AuthorizationConfig.ClientConfig(getValueAsString(this, "private-key")!!)
|
||||
}
|
||||
|
||||
if (authClient == null && authServer == null) {
|
||||
throw IllegalStateException("Token auth server settings are not specified")
|
||||
}
|
||||
|
||||
return AuthorizationConfig(enabled, publicKeyOwner, privateKey, publicKey)
|
||||
return AuthorizationConfig(
|
||||
enabled, publicKeyOwner,
|
||||
authServer ?: AuthorizationConfig.ServerConfig.default(),
|
||||
authClient ?: AuthorizationConfig.ClientConfig.default()
|
||||
)
|
||||
}
|
||||
|
||||
private fun fileNotExists(path: String): Boolean {
|
||||
|
||||
@@ -150,6 +150,7 @@ open class UpstreamsConfig {
|
||||
var host: String? = null
|
||||
var port: Int = 0
|
||||
var auth: AuthConfig.ClientTlsAuth? = null
|
||||
var tokenAuth: AuthConfig.ClientTokenAuth? = null
|
||||
var upstreamRating: Int = 0
|
||||
}
|
||||
|
||||
|
||||
@@ -114,6 +114,7 @@ class UpstreamsConfigReader(
|
||||
connection.port = it
|
||||
}
|
||||
connection.auth = authConfigReader.readClientTls(connConfigNode)
|
||||
connection.tokenAuth = authConfigReader.readTokenAuth(connConfigNode)
|
||||
} else {
|
||||
log.error("Upstream at #0 has invalid configuration")
|
||||
}
|
||||
|
||||
@@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.Chain
|
||||
import io.emeraldpay.dshackle.Defaults
|
||||
import io.emeraldpay.dshackle.FileResolver
|
||||
import io.emeraldpay.dshackle.Global
|
||||
import io.emeraldpay.dshackle.config.AuthorizationConfig
|
||||
import io.emeraldpay.dshackle.config.ChainsConfig
|
||||
import io.emeraldpay.dshackle.config.CompressionConfig
|
||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||
@@ -51,6 +52,7 @@ import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
|
||||
import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
|
||||
import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
|
||||
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams
|
||||
import io.emeraldpay.dshackle.upstream.grpc.auth.GrpcAuthContext
|
||||
import io.grpc.ClientInterceptor
|
||||
import org.slf4j.LoggerFactory
|
||||
import org.springframework.beans.factory.annotation.Autowired
|
||||
@@ -84,7 +86,9 @@ open class ConfiguredUpstreams(
|
||||
@Autowired(required = false)
|
||||
private val clientSpansInterceptor: ClientInterceptor?,
|
||||
@Qualifier("headScheduler")
|
||||
private val headScheduler: Scheduler
|
||||
private val headScheduler: Scheduler,
|
||||
private val authorizationConfig: AuthorizationConfig,
|
||||
private val grpcAuthContext: GrpcAuthContext
|
||||
) : ApplicationRunner {
|
||||
@Value("\${spring.application.max-metadata-size}")
|
||||
private var maxMetadataSize: Int = Defaults.maxMetadataSize
|
||||
@@ -351,6 +355,8 @@ open class ConfiguredUpstreams(
|
||||
endpoint.host!!,
|
||||
endpoint.port,
|
||||
endpoint.auth,
|
||||
endpoint.tokenAuth,
|
||||
authorizationConfig,
|
||||
compression,
|
||||
fileResolver,
|
||||
endpoint.upstreamRating,
|
||||
@@ -361,7 +367,8 @@ open class ConfiguredUpstreams(
|
||||
grpcTracing,
|
||||
clientSpansInterceptor,
|
||||
maxMetadataSize,
|
||||
headScheduler
|
||||
headScheduler,
|
||||
grpcAuthContext
|
||||
).apply {
|
||||
timeout = options.timeout
|
||||
}
|
||||
|
||||
@@ -22,22 +22,30 @@ import io.emeraldpay.api.proto.BlockchainOuterClass.DescribeResponse
|
||||
import io.emeraldpay.api.proto.BlockchainOuterClass.StatusRequest
|
||||
import io.emeraldpay.api.proto.Common
|
||||
import io.emeraldpay.api.proto.Common.ChainRef.UNRECOGNIZED
|
||||
import io.emeraldpay.api.proto.ReactorAuthGrpc
|
||||
import io.emeraldpay.api.proto.ReactorBlockchainGrpc
|
||||
import io.emeraldpay.dshackle.BlockchainType
|
||||
import io.emeraldpay.dshackle.Chain
|
||||
import io.emeraldpay.dshackle.Defaults
|
||||
import io.emeraldpay.dshackle.FileResolver
|
||||
import io.emeraldpay.dshackle.config.AuthConfig
|
||||
import io.emeraldpay.dshackle.config.AuthorizationConfig
|
||||
import io.emeraldpay.dshackle.config.ChainsConfig
|
||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||
import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
|
||||
import io.emeraldpay.dshackle.upstream.DefaultUpstream
|
||||
import io.emeraldpay.dshackle.upstream.Lifecycle
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
|
||||
import io.emeraldpay.dshackle.upstream.grpc.auth.AuthException
|
||||
import io.emeraldpay.dshackle.upstream.grpc.auth.ClientAuthenticationInterceptor
|
||||
import io.emeraldpay.dshackle.upstream.grpc.auth.GrpcAuthContext
|
||||
import io.emeraldpay.dshackle.upstream.grpc.auth.GrpcUpstreamsAuth
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
|
||||
import io.grpc.ClientInterceptor
|
||||
import io.grpc.Codec
|
||||
import io.grpc.Status
|
||||
import io.grpc.StatusRuntimeException
|
||||
import io.grpc.netty.NettyChannelBuilder
|
||||
import io.micrometer.core.instrument.Counter
|
||||
import io.micrometer.core.instrument.Metrics
|
||||
@@ -52,6 +60,7 @@ import org.apache.commons.lang3.exception.ExceptionUtils
|
||||
import org.slf4j.LoggerFactory
|
||||
import reactor.core.Disposable
|
||||
import reactor.core.publisher.Flux
|
||||
import reactor.core.publisher.Mono
|
||||
import reactor.core.scheduler.Scheduler
|
||||
import java.io.IOException
|
||||
import java.time.Duration
|
||||
@@ -66,6 +75,8 @@ class GrpcUpstreams(
|
||||
private val host: String,
|
||||
private val port: Int,
|
||||
private val auth: AuthConfig.ClientTlsAuth? = null,
|
||||
private val tokenAuth: AuthConfig.ClientTokenAuth? = null,
|
||||
private val authorizationConfig: AuthorizationConfig,
|
||||
private val compression: Boolean,
|
||||
private val fileResolver: FileResolver,
|
||||
private val nodeRating: Int,
|
||||
@@ -76,7 +87,8 @@ class GrpcUpstreams(
|
||||
private val grpcTracing: GrpcTracing,
|
||||
private val clientSpansInterceptor: ClientInterceptor?,
|
||||
private var maxMetadataSize: Int,
|
||||
private val headScheduler: Scheduler
|
||||
private val headScheduler: Scheduler,
|
||||
private val grpcAuthContext: GrpcAuthContext
|
||||
) {
|
||||
private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java)
|
||||
|
||||
@@ -92,7 +104,10 @@ class GrpcUpstreams(
|
||||
.maxInboundMessageSize(Defaults.maxMessageSize)
|
||||
.maxInboundMetadataSize(maxMetadataSize)
|
||||
.enableRetry()
|
||||
.intercept(grpcTracing.newClientInterceptor())
|
||||
.intercept(
|
||||
grpcTracing.newClientInterceptor(),
|
||||
ClientAuthenticationInterceptor(id, grpcAuthContext),
|
||||
)
|
||||
.executor(grpcExecutor)
|
||||
.maxRetryAttempts(3)
|
||||
clientSpansInterceptor?.let {
|
||||
@@ -108,17 +123,28 @@ class GrpcUpstreams(
|
||||
chanelBuilder.usePlaintext()
|
||||
}
|
||||
|
||||
var client = ReactorBlockchainGrpc.newReactorStub(chanelBuilder.build())
|
||||
val channel = chanelBuilder.build()
|
||||
var client = ReactorBlockchainGrpc.newReactorStub(channel)
|
||||
if (compression) {
|
||||
client = client.withCompression(Codec.Gzip().messageEncoding)
|
||||
}
|
||||
this.client = client
|
||||
|
||||
val grpcUpstreamsAuth =
|
||||
if (tokenAuth != null && authorizationConfig.enabled) {
|
||||
GrpcUpstreamsAuth(
|
||||
ReactorAuthGrpc.newReactorStub(channel),
|
||||
authorizationConfig,
|
||||
grpcAuthContext,
|
||||
tokenAuth.publicKeyPath!!
|
||||
)
|
||||
} else null
|
||||
|
||||
val statusSubscriptions = mutableMapOf<Chain, Disposable>()
|
||||
|
||||
return Flux.interval(Duration.ZERO, Duration.ofSeconds(20))
|
||||
.flatMap {
|
||||
client.describe(DescribeRequest.newBuilder().build())
|
||||
authAndDescribe(grpcUpstreamsAuth)
|
||||
}.onErrorContinue { t, _ ->
|
||||
if (ExceptionUtils.indexOfType(t, IOException::class.java) >= 0) {
|
||||
log.warn("gRPC upstream $host:$port is unavailable. (${t.javaClass}: ${t.message})")
|
||||
@@ -133,7 +159,7 @@ class GrpcUpstreams(
|
||||
}.doOnNext {
|
||||
val sub = statusSubscriptions[it.chain]
|
||||
if (sub == null || sub.isDisposed) {
|
||||
val subscription = client.subscribeStatus(
|
||||
val subscription = this.client.subscribeStatus(
|
||||
StatusRequest.newBuilder()
|
||||
.addChains(Common.ChainRef.forNumber(it.chain.id)).build()
|
||||
).subscribeOn(chainStatusScheduler)
|
||||
@@ -293,4 +319,43 @@ class GrpcUpstreams(
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun authAndDescribe(grpcUpstreamsAuth: GrpcUpstreamsAuth?): Mono<DescribeResponse> {
|
||||
return Mono.justOrEmpty(grpcUpstreamsAuth)
|
||||
.flatMap {
|
||||
if (grpcAuthContext.containsToken(id)) {
|
||||
Mono.empty()
|
||||
} else {
|
||||
auth(it)
|
||||
}
|
||||
}
|
||||
.then(
|
||||
describe()
|
||||
.onErrorResume {
|
||||
if (it is StatusRuntimeException && it.status.code == Status.Code.UNAUTHENTICATED) {
|
||||
grpcAuthContext.removeToken(id)
|
||||
auth(grpcUpstreamsAuth).then(describe())
|
||||
} else {
|
||||
Mono.error(it)
|
||||
}
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
private fun auth(grpcUpstreamsAuth: GrpcUpstreamsAuth?): Mono<Void> {
|
||||
if (grpcUpstreamsAuth == null) {
|
||||
return Mono.empty()
|
||||
}
|
||||
return grpcUpstreamsAuth.auth(id)
|
||||
.flatMap { authRes ->
|
||||
if (!authRes.passed) {
|
||||
log.warn(authRes.cause)
|
||||
Mono.error<AuthException>(AuthException(authRes.cause!!))
|
||||
} else {
|
||||
Mono.empty()
|
||||
}
|
||||
}.then()
|
||||
}
|
||||
|
||||
private fun describe() = this.client.describe(DescribeRequest.newBuilder().build())
|
||||
}
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
package io.emeraldpay.dshackle.upstream.grpc.auth
|
||||
|
||||
import io.emeraldpay.dshackle.auth.processor.SESSION_ID
|
||||
import io.grpc.CallOptions
|
||||
import io.grpc.Channel
|
||||
import io.grpc.ClientCall
|
||||
import io.grpc.ClientInterceptor
|
||||
import io.grpc.ForwardingClientCall
|
||||
import io.grpc.Metadata
|
||||
import io.grpc.MethodDescriptor
|
||||
|
||||
class ClientAuthenticationInterceptor(
|
||||
private val upstreamId: String,
|
||||
private val grpcAuthContext: GrpcAuthContext
|
||||
) : ClientInterceptor {
|
||||
|
||||
companion object {
|
||||
private val AUTHORIZATION_HEADER: Metadata.Key<String> =
|
||||
Metadata.Key.of(SESSION_ID, Metadata.ASCII_STRING_MARSHALLER)
|
||||
}
|
||||
|
||||
override fun <ReqT, RespT> interceptCall(
|
||||
method: MethodDescriptor<ReqT, RespT>,
|
||||
callOptions: CallOptions,
|
||||
next: Channel
|
||||
): ClientCall<ReqT, RespT> =
|
||||
object : ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
|
||||
override fun start(responseListener: Listener<RespT>, headers: Metadata) {
|
||||
grpcAuthContext.getToken(upstreamId)?.let {
|
||||
headers.put(AUTHORIZATION_HEADER, it)
|
||||
}
|
||||
super.start(responseListener, headers)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
package io.emeraldpay.dshackle.upstream.grpc.auth
|
||||
|
||||
import org.springframework.stereotype.Component
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
@Component
|
||||
class GrpcAuthContext {
|
||||
private val sessions = ConcurrentHashMap<String, String>()
|
||||
|
||||
fun putTokenInContext(upstreamId: String, sessionId: String) {
|
||||
sessions[upstreamId] = sessionId
|
||||
}
|
||||
|
||||
fun removeToken(upstreamId: String) {
|
||||
sessions.remove(upstreamId)
|
||||
}
|
||||
|
||||
fun containsToken(upstreamId: String) = sessions.containsKey(upstreamId)
|
||||
|
||||
fun getToken(upstreamId: String) = sessions[upstreamId]
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
package io.emeraldpay.dshackle.upstream.grpc.auth
|
||||
|
||||
import com.auth0.jwt.JWT
|
||||
import com.auth0.jwt.JWTVerifier
|
||||
import com.auth0.jwt.algorithms.Algorithm
|
||||
import io.emeraldpay.api.proto.AuthOuterClass
|
||||
import io.emeraldpay.api.proto.ReactorAuthGrpc.ReactorAuthStub
|
||||
import io.emeraldpay.dshackle.auth.processor.AuthVersion
|
||||
import io.emeraldpay.dshackle.auth.processor.SESSION_ID
|
||||
import io.emeraldpay.dshackle.auth.processor.VERSION
|
||||
import io.emeraldpay.dshackle.auth.service.RsaKeyReader
|
||||
import io.emeraldpay.dshackle.config.AuthorizationConfig
|
||||
import reactor.core.publisher.Mono
|
||||
import java.security.interfaces.RSAPrivateKey
|
||||
import java.security.interfaces.RSAPublicKey
|
||||
import java.time.Instant
|
||||
|
||||
class AuthException(message: String) : RuntimeException(message)
|
||||
|
||||
class GrpcUpstreamsAuth(
|
||||
private val authClient: ReactorAuthStub,
|
||||
private val authorizationConfig: AuthorizationConfig,
|
||||
private val grpcAuthContext: GrpcAuthContext,
|
||||
publicKeyPath: String
|
||||
) {
|
||||
private val rsaKeyReader = RsaKeyReader()
|
||||
private val keys = rsaKeyReader.getKeyPair(authorizationConfig.clientConfig.privateKeyPath, publicKeyPath)
|
||||
|
||||
fun auth(providerId: String): Mono<AuthResult> {
|
||||
return authClient.authenticate(
|
||||
AuthOuterClass.AuthRequest.newBuilder()
|
||||
.setToken(generateToken())
|
||||
.build()
|
||||
).map {
|
||||
verify(it.providerToken, providerId)
|
||||
}.onErrorResume {
|
||||
Mono.just(AuthResult(false, "Error during auth - ${it.message}"))
|
||||
}
|
||||
}
|
||||
|
||||
private fun generateToken(): String {
|
||||
return JWT.create()
|
||||
.withIssuedAt(Instant.now())
|
||||
.withIssuer(authorizationConfig.publicKeyOwner)
|
||||
.withClaim(VERSION, AuthVersion.V1.toString())
|
||||
.sign(Algorithm.RSA256(keys.providerPrivateKey as RSAPrivateKey))
|
||||
}
|
||||
|
||||
private fun verify(token: String, providerId: String): AuthResult {
|
||||
val verifier: JWTVerifier = JWT
|
||||
.require(Algorithm.RSA256(keys.externalPublicKey as RSAPublicKey, null))
|
||||
.withClaim(SESSION_ID) { claim, _ -> !claim.isMissing }
|
||||
.build()
|
||||
val decodedToken = verifier.verify(token)
|
||||
grpcAuthContext.putTokenInContext(providerId, decodedToken.getClaim(SESSION_ID).asString())
|
||||
|
||||
return AuthResult(true)
|
||||
}
|
||||
|
||||
data class AuthResult(
|
||||
val passed: Boolean,
|
||||
val cause: String? = null
|
||||
)
|
||||
}
|
||||
Reference in New Issue
Block a user