From 08826ecbc4acdd48da6bcf22e2051f5f9fa8ab12 Mon Sep 17 00:00:00 2001 From: Maxksim Fomenkov Date: Wed, 14 Sep 2022 10:14:34 +0300 Subject: [PATCH] added signature for native subscribe --- dshackle-cli/grpc | 2 +- emerald-java-client | 2 +- .../dshackle/data/BlockContainer.kt | 34 +++++++----- .../dshackle/quorum/QuorumRpcReader.kt | 2 +- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 2 +- .../dshackle/rpc/NativeSubscribe.kt | 52 ++++++++++++++++--- .../upstream/ethereum/DefaultEthereumHead.kt | 3 +- .../upstream/ethereum/EthereumDirectReader.kt | 2 +- .../upstream/ethereum/EthereumRpcHead.kt | 3 +- .../upstream/ethereum/EthereumWsFactory.kt | 4 +- .../upstream/ethereum/EthereumWsHead.kt | 3 +- .../upstream/ethereum/WsConnection.kt | 5 +- .../connectors/EthereumConnectorFactory.kt | 6 ++- .../connectors/EthereumRpcConnector.kt | 8 +-- .../connectors/EthereumWsConnector.kt | 5 +- .../ethereum/subscribe/ConnectBlockUpdates.kt | 11 ++-- .../ethereum/subscribe/ProduceLogs.kt | 4 +- .../ethereum/subscribe/ProduceNewHeads.kt | 19 ++++--- .../ethereum/subscribe/json/HasUpstream.kt | 5 ++ .../ethereum/subscribe/json/LogMessage.kt | 7 ++- .../ethereum/subscribe/json/NewHeadMessage.kt | 7 ++- .../upstream/grpc/EthereumGrpcUpstream.kt | 2 +- .../upstream/grpc/EthereumPosGrpcUpstream.kt | 2 +- .../upstream/signature/EcdsaSigner.kt | 10 ++-- .../dshackle/upstream/signature/NoSigner.kt | 2 +- .../upstream/signature/ResponseSigner.kt | 2 +- .../upstream/signature/EcdsaSignerSpec.groovy | 4 +- 27 files changed, 137 insertions(+), 71 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/HasUpstream.kt diff --git a/dshackle-cli/grpc b/dshackle-cli/grpc index 0d32b45d..48ea13c0 160000 --- a/dshackle-cli/grpc +++ b/dshackle-cli/grpc @@ -1 +1 @@ -Subproject commit 0d32b45d007417c901ca704c9324b5e6c7167f4f +Subproject commit 48ea13c0d1ec8498d249039bc815a4698585873a diff --git a/emerald-java-client b/emerald-java-client index a3dde262..b05afe6a 160000 --- a/emerald-java-client +++ b/emerald-java-client @@ -1 +1 @@ -Subproject commit a3dde262ffab10749b8f14532765e0d2054fbd98 +Subproject commit b05afe6af1008c2cb2c04838c91f4c0ff05589b9 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt b/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt index ac411dbb..52404ccb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt @@ -31,34 +31,40 @@ class BlockContainer( json: ByteArray?, val parsed: Any?, val transactions: List = emptyList(), - val nodeRating: Int = 0 + val nodeRating: Int = 0, + val upstreamId: String = "" ) : SourceContainer(json, parsed) { companion object { @JvmStatic - fun from(block: BlockJson<*>, raw: ByteArray): BlockContainer { - val hasTransactions = block.transactions?.filterIsInstance()?.count() ?: 0 > 0 + fun from(block: BlockJson<*>, raw: ByteArray, upstreamId: String): BlockContainer { + val hasTransactions = !block.transactions?.filterIsInstance().isNullOrEmpty() return BlockContainer( - block.number, - BlockId.from(block), - block.totalDifficulty, - block.timestamp, - hasTransactions, - raw, - block, - block.transactions?.map { TxId.from(it.hash) } ?: emptyList() + height = block.number, + hash = BlockId.from(block), + difficulty = block.totalDifficulty, + timestamp = block.timestamp, + full = hasTransactions, + json = raw, + parsed = block, + transactions = block.transactions?.map { TxId.from(it.hash) } ?: emptyList(), + upstreamId = upstreamId ) } @JvmStatic fun from(block: BlockJson<*>): BlockContainer { - return from(block, Global.objectMapper.writeValueAsBytes(block)) + return from(block, "unknown") + } + @JvmStatic + fun from(block: BlockJson<*>, upstream: String): BlockContainer { + return from(block, Global.objectMapper.writeValueAsBytes(block), upstream) } @JvmStatic - fun fromEthereumJson(raw: ByteArray): BlockContainer { + fun fromEthereumJson(raw: ByteArray, upstream: String): BlockContainer { val block = Global.objectMapper.readValue(raw, BlockJson::class.java) - return from(block, raw) + return from(block, raw, upstream) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt index fbe72bf4..56cc9a94 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt @@ -141,7 +141,7 @@ class QuorumRpcReader( src.map { val signature = response.providedSignature ?: if (key.nonce != null) { - signer?.sign(key.nonce, response.getResult(), api) + signer?.sign(key.nonce, response.getResult(), api.getId()) } else { null } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index bcb3dce0..a7b3dab9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -220,7 +220,7 @@ open class NativeCall( .flatMap(JsonRpcResponse::requireResult) .map { if (ctx.nonce != null) { - CallResult.ok(ctx.id, ctx.nonce, it, signer.sign(ctx.nonce, it, ctx.upstream)) + CallResult.ok(ctx.id, ctx.nonce, it, signer.sign(ctx.nonce, it, ctx.upstream.getId())) } else { CallResult.ok(ctx.id, null, it, null) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt index 1169b03b..ce253fa7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt @@ -22,6 +22,8 @@ import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.upstream.MultistreamHolder import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream +import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.HasUpstream +import io.emeraldpay.dshackle.upstream.signature.ResponseSigner import io.emeraldpay.grpc.BlockchainType import io.emeraldpay.grpc.Chain import io.grpc.Status @@ -35,7 +37,8 @@ import reactor.core.publisher.Mono @Service open class NativeSubscribe( - @Autowired private val multistreamHolder: MultistreamHolder + @Autowired private val multistreamHolder: MultistreamHolder, + @Autowired private val signer: ResponseSigner ) { companion object { @@ -51,7 +54,7 @@ open class NativeSubscribe( .onErrorMap(this@NativeSubscribe::convertToStatus) } - fun start(request: BlockchainOuterClass.NativeSubscribeRequest): Publisher { + fun start(request: BlockchainOuterClass.NativeSubscribeRequest): Publisher { val chain = Chain.byId(request.chainValue) if (BlockchainType.from(chain) != BlockchainType.ETHEREUM_POS && BlockchainType.from(chain) != BlockchainType.ETHEREUM) { return Mono.error(UnsupportedOperationException("Native subscribe is not supported for ${chain.chainCode}")) @@ -61,7 +64,9 @@ open class NativeSubscribe( objectMapper.readValue(it.newInput(), Map::class.java) } val matcher = Selector.convertToMatcher(request.selector) - return subscribe(chain, method, params, matcher) + return subscribe(chain, method, params, matcher).map { resp -> + ResponseHolder(resp, request.nonce.takeIf { it != 0L }) + } } fun convertToStatus(t: Throwable) = when (t) { @@ -86,10 +91,43 @@ open class NativeSubscribe( .subscribe(method, params, matcher) } - fun convertToProto(value: Any): BlockchainOuterClass.NativeSubscribeReplyItem { - val result = objectMapper.writeValueAsBytes(value) - return BlockchainOuterClass.NativeSubscribeReplyItem.newBuilder() + fun convertToProto(holder: ResponseHolder): BlockchainOuterClass.NativeSubscribeReplyItem { + val result = objectMapper.writeValueAsBytes(holder.response) + val builder = BlockchainOuterClass.NativeSubscribeReplyItem.newBuilder() .setPayload(ByteString.copyFrom(result)) - .build() + + holder.nonce?.also { nonce -> + holder.getSource()?.let { + signer.sign(nonce, result, it) + }?.let { + buildSignature(nonce, it) + }?.also { + builder.signature = it + } + } + + return builder.build() + } + + fun buildSignature( + nonce: Long, + signature: ResponseSigner.Signature + ): BlockchainOuterClass.NativeCallReplySignature { + val msg = BlockchainOuterClass.NativeCallReplySignature.newBuilder() + msg.signature = ByteString.copyFrom(signature.value) + msg.keyId = signature.keyId + msg.upstreamId = signature.upstreamId + msg.nonce = nonce + return msg.build() + } + + data class ResponseHolder( + val response: Any, + val nonce: Long? + ) { + fun getSource(): String? = + if (response is HasUpstream) { + response.upstreamId.takeIf { it != "unknown" } + } else null } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt index eee280c0..66b2f129 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt @@ -29,6 +29,7 @@ import org.slf4j.LoggerFactory import reactor.core.publisher.Mono open class DefaultEthereumHead( + private val upstreamId: String, forkChoice: ForkChoice, blockValidator: BlockValidator ) : Head, AbstractHead(forkChoice, blockValidator) { @@ -57,7 +58,7 @@ open class DefaultEthereumHead( .timeout(Defaults.timeout, Mono.error(Exception("Block data not received"))) } .map { - BlockContainer.fromEthereumJson(it.getResult()) + BlockContainer.fromEthereumJson(it.getResult(), upstreamId) } .onErrorResume { err -> log.debug("Failed to fetch latest block: ${err.message}") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt index 06ec5921..15bc299c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -119,7 +119,7 @@ class EthereumDirectReader( if (block == null) { Mono.empty() } else { - Mono.just(BlockContainer.from(block, blockbytes)) + Mono.just(BlockContainer.from(block, blockbytes, "unknown")) } } .doOnNext { block -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt index bb1fb38d..d299354f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt @@ -33,9 +33,10 @@ import java.util.concurrent.Executors class EthereumRpcHead( private val api: Reader, forkChoice: ForkChoice, + upstreamId: String, blockValidator: BlockValidator, private val interval: Duration = Duration.ofSeconds(10), -) : DefaultEthereumHead(forkChoice, blockValidator), Lifecycle { +) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator), Lifecycle { companion object { val scheduler = diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt index 00def0bf..d81cf822 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -58,11 +58,11 @@ class EthereumWsFactory( ) } - fun create(upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?): WsConnection { + fun create(id: String, upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?): WsConnection { require(upstream == null || upstream.getId() == id) { "Creating instance for different upstream. ${upstream?.getId()} != id" } - return WsConnection(uri, origin, basicAuth, metrics, upstream, validator).also { ws -> + return WsConnection(id, uri, origin, basicAuth, metrics, upstream, validator).also { ws -> config?.frameSize?.let { ws.frameSize = it } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt index d008530f..e34ef409 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt @@ -26,9 +26,10 @@ import reactor.core.publisher.Flux class EthereumWsHead( private val ws: WsConnection, + upstreamId: String, forkChoice: ForkChoice, blockValidator: BlockValidator -) : DefaultEthereumHead(forkChoice, blockValidator), Lifecycle { +) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator), Lifecycle { private val log = LoggerFactory.getLogger(EthereumWsHead::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt index a1c95ff8..b2b6bec2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt @@ -65,6 +65,7 @@ import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger open class WsConnection( + private val id: String, private val uri: URI, private val origin: URI, private val basicAuth: AuthConfig.ClientBasicAuth?, @@ -352,7 +353,7 @@ open class WsConnection( } } .flatMap(JsonRpcResponse::requireResult) - .map { BlockContainer.fromEthereumJson(it) } + .map { BlockContainer.fromEthereumJson(it, id) } .subscribeOn(Schedulers.boundedElastic()) .timeout(Defaults.timeoutInternal, Mono.empty()) }.repeatWhenEmpty { n -> @@ -368,7 +369,7 @@ open class WsConnection( .then() } else { Mono.fromCallable { - blocks.tryEmitNext(BlockContainer.from(block)) + blocks.tryEmitNext(BlockContainer.from(block, id)) }.then() } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt index f7ef9b36..f23efb6f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt @@ -16,7 +16,9 @@ open class EthereumConnectorFactory( private val forkChoice: ForkChoice, private val blockValidator: BlockValidator ) : ConnectorFactory { - private val log = LoggerFactory.getLogger(EthereumConnectorFactory::class.java) + companion object { + private val log = LoggerFactory.getLogger(EthereumConnectorFactory::class.java) + } override fun isValid(): Boolean { if (preferHttp && httpFactory == null) { @@ -27,7 +29,7 @@ open class EthereumConnectorFactory( override fun create(upstream: DefaultUpstream, validator: EthereumUpstreamValidator, chain: Chain): EthereumConnector { if (wsFactory != null && !preferHttp) { - return EthereumWsConnector(wsFactory, upstream, validator, chain, forkChoice, blockValidator) + return EthereumWsConnector(wsFactory, upstream, validator, forkChoice, blockValidator) } if (httpFactory == null) { throw java.lang.IllegalArgumentException("Can't create rpc connector if no http factory set") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt index 8ee56bbe..61a61704 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt @@ -34,15 +34,15 @@ class EthereumRpcConnector( init { if (wsFactory != null) { // do not set upstream to the WS, since it doesn't control the RPC upstream - conn = wsFactory.create(null, null) - val wsHead = EthereumWsHead(conn, forkChoice, blockValidator) + conn = wsFactory.create(id, null, null) + val wsHead = EthereumWsHead(conn, id, forkChoice, blockValidator) // receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed - val rpcHead = EthereumRpcHead(directReader, forkChoice, blockValidator, Duration.ofSeconds(60)) + val rpcHead = EthereumRpcHead(directReader, forkChoice, id, blockValidator, Duration.ofSeconds(60)) head = MergedHead(listOf(rpcHead, wsHead), forkChoice) } else { conn = null log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC") - head = EthereumRpcHead(directReader, forkChoice, blockValidator) + head = EthereumRpcHead(directReader, forkChoice, id, blockValidator) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt index eb796104..83fde6b8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt @@ -15,7 +15,6 @@ class EthereumWsConnector( wsFactory: EthereumWsFactory, upstream: DefaultUpstream, validator: EthereumUpstreamValidator, - chain: Chain, forkChoice: ForkChoice, blockValidator: BlockValidator ) : EthereumConnector { @@ -24,8 +23,8 @@ class EthereumWsConnector( private val head: EthereumWsHead init { - conn = wsFactory.create(upstream, validator) - head = EthereumWsHead(conn, forkChoice, blockValidator) + conn = wsFactory.create(upstream.getId(), upstream, validator) + head = EthereumWsHead(conn, upstream.getId(), forkChoice, blockValidator) api = JsonRpcWsClient(conn) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt index 7d635b62..87ed40cf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt @@ -71,7 +71,7 @@ class ConnectBlockUpdates( val prev = findPrevious(block) remember(block) val removed = if (prev != null) { - whenReplaced(prev) + whenReplaced(prev, block.upstreamId) } else { Flux.empty() } @@ -103,13 +103,14 @@ class ConnectBlockUpdates( /** * Produce updates for transactions when a block is replaces with a different one on the same height. */ - fun whenReplaced(prev: BlockContainer): Flux { + fun whenReplaced(prev: BlockContainer, source: String): Flux { return Flux.fromIterable(prev.transactions).map { Update( prev.hash, prev.height, UpdateType.DROP, - it + it, + source ) } } @@ -121,7 +122,8 @@ class ConnectBlockUpdates( block.hash, block.height, UpdateType.NEW, - it + it, + block.upstreamId ) } } @@ -131,6 +133,7 @@ class ConnectBlockUpdates( val blockNumber: Long, val type: UpdateType, val transactionId: TxId, + val upstreamId: String ) enum class UpdateType { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt index 44da804d..a7c11ba6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt @@ -94,7 +94,9 @@ class ProduceLogs( txlog.topics, txlog.transactionHash, txlog.transactionIndex, - false + false, + update.upstreamId + ) } oldMessages.put(LogReference(update.blockHash, update.transactionId), messages) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceNewHeads.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceNewHeads.kt index 3b9f6162..68fa7d22 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceNewHeads.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceNewHeads.kt @@ -16,6 +16,7 @@ package io.emeraldpay.dshackle.upstream.ethereum.subscribe import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHeadMessage import io.emeraldpay.etherjar.rpc.json.BlockJson @@ -41,13 +42,7 @@ class ProduceNewHeads( fun start(): Flux { return head.getFlux() .map { - if (it.parsed != null) { - it.parsed as BlockJson - } else { - objectMapper.readValue(it.json, BlockJson::class.java) - } - } - .map { block -> + val block = extractBlock(it) NewHeadMessage( block.number, block.hash, @@ -58,8 +53,16 @@ class ProduceNewHeads( block.gasUsed, block.logsBloom, block.miner, - block.baseFeePerGas?.amount + block.baseFeePerGas?.amount, + it.upstreamId ) } } + + private fun extractBlock(blockContainer: BlockContainer): BlockJson = + if (blockContainer.parsed != null) { + blockContainer.parsed as BlockJson + } else { + objectMapper.readValue(blockContainer.json, BlockJson::class.java) + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/HasUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/HasUpstream.kt new file mode 100644 index 00000000..fcbe2185 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/HasUpstream.kt @@ -0,0 +1,5 @@ +package io.emeraldpay.dshackle.upstream.ethereum.subscribe.json + +interface HasUpstream { + val upstreamId: String +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/LogMessage.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/LogMessage.kt index f0ef7d53..88c77530 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/LogMessage.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/LogMessage.kt @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.upstream.ethereum.subscribe.json +import com.fasterxml.jackson.annotation.JsonIgnore import com.fasterxml.jackson.databind.annotation.JsonSerialize import io.emeraldpay.etherjar.domain.Address import io.emeraldpay.etherjar.domain.BlockHash @@ -40,5 +41,7 @@ data class LogMessage( val transactionHash: TransactionId, @get:JsonSerialize(using = NumberAsHexSerializer::class) val transactionIndex: Long, - val removed: Boolean -) + val removed: Boolean, + @get:JsonIgnore + override val upstreamId: String +) : HasUpstream diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadMessage.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadMessage.kt index 90eec052..5dd5b175 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadMessage.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadMessage.kt @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.upstream.ethereum.subscribe.json +import com.fasterxml.jackson.annotation.JsonIgnore import com.fasterxml.jackson.annotation.JsonInclude import com.fasterxml.jackson.databind.annotation.JsonSerialize import io.emeraldpay.etherjar.domain.Address @@ -50,5 +51,7 @@ data class NewHeadMessage( val miner: Address, @get:JsonSerialize(using = NumberAsHexSerializer::class) @get:JsonInclude(JsonInclude.Include.NON_NULL) - val baseFeePerGas: BigInteger? -) + val baseFeePerGas: BigInteger?, + @get:JsonIgnore + override val upstreamId: String +) : HasUpstream diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index 7f79bf95..1459550a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -84,7 +84,7 @@ open class EthereumGrpcUpstream( defaultReader.read(JsonRpcRequest("eth_getBlockByHash", listOf(existingBlock.hash.toHexWithPrefix(), false))) .flatMap(JsonRpcResponse::requireResult) .map { - BlockContainer.fromEthereumJson(it) + BlockContainer.fromEthereumJson(it, getId()) } .timeout(timeout, Mono.error(TimeoutException("Timeout from upstream"))) .doOnError { t -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt index 256f0a4f..31f4c288 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt @@ -84,7 +84,7 @@ open class EthereumPosGrpcUpstream( defaultReader.read(JsonRpcRequest("eth_getBlockByHash", listOf(existingBlock.hash.toHexWithPrefix(), false))) .flatMap(JsonRpcResponse::requireResult) .map { - BlockContainer.fromEthereumJson(it) + BlockContainer.fromEthereumJson(it, getId()) } .timeout(timeout, Mono.error(TimeoutException("Timeout from upstream"))) .doOnError { t -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/EcdsaSigner.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/EcdsaSigner.kt index 10039c13..54e7507d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/EcdsaSigner.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/EcdsaSigner.kt @@ -17,15 +17,13 @@ class EcdsaSigner( const val MSG_SEPARATOR = '/' } - override fun sign(nonce: Long, message: ByteArray, source: Upstream): ResponseSigner.Signature { + override fun sign(nonce: Long, message: ByteArray, source: String): ResponseSigner.Signature { val sig = Signature.getInstance(SIGN_SCHEME) sig.initSign(privateKey) val wrapped = wrapMessage(nonce, message, source) sig.update(wrapped.toByteArray()) val value = sig.sign() - return ResponseSigner.Signature( - value, source.getId(), keyId - ) + return ResponseSigner.Signature(value, source, keyId) } /** @@ -42,7 +40,7 @@ class EcdsaSigner( * - second is the nonce value encode as decimal string * - third is SHA256 hash of the original message encoded as hex string */ - fun wrapMessage(nonce: Long, message: ByteArray, source: Upstream): String { + fun wrapMessage(nonce: Long, message: ByteArray, source: String): String { val sha256 = MessageDigest.getInstance("SHA-256") // we create it with max capacity that we expect for the result, which is total lengths of its parts val formatterMsg = StringBuilder(11 + 1 + 18 + 1 + 64 + 1 + 64) @@ -52,7 +50,7 @@ class EcdsaSigner( .append(MSG_SEPARATOR) // We expect that the id is short enough (less than 64 symbols) and also it doesn't contain the `/` symbol // which is verified in UpstreamConfigReader and DefaultUpstream constructor - .append(source.getId()) + .append(source) .append(MSG_SEPARATOR) .append(Hex.encodeHexString(sha256.digest(message))) return formatterMsg.toString() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/NoSigner.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/NoSigner.kt index 2945041d..c1c4f522 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/NoSigner.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/NoSigner.kt @@ -3,7 +3,7 @@ package io.emeraldpay.dshackle.upstream.signature import io.emeraldpay.dshackle.upstream.Upstream class NoSigner : ResponseSigner { - override fun sign(nonce: Long, message: ByteArray, source: Upstream): ResponseSigner.Signature? { + override fun sign(nonce: Long, message: ByteArray, source: String): ResponseSigner.Signature? { return null } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/ResponseSigner.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/ResponseSigner.kt index d88aee4e..057cea55 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/ResponseSigner.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/ResponseSigner.kt @@ -4,7 +4,7 @@ import io.emeraldpay.dshackle.upstream.Upstream interface ResponseSigner { - fun sign(nonce: Long, message: ByteArray, source: Upstream): Signature? + fun sign(nonce: Long, message: ByteArray, source: String): Signature? data class Signature( val value: ByteArray, diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/signature/EcdsaSignerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/signature/EcdsaSignerSpec.groovy index 7368290d..13a32cfe 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/signature/EcdsaSignerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/signature/EcdsaSignerSpec.groovy @@ -118,7 +118,7 @@ class EcdsaSignerSpec extends Specification { def signer = new EcdsaSigner((pair.getPrivate() as ECPrivateKey), 100L) when: - def sig = signer.sign(10, result, up) + def sig = signer.sign(10, result, up.id) then: verifier.verify(sig.value) @@ -148,7 +148,7 @@ class EcdsaSignerSpec extends Specification { def signer = factory.getObject() as EcdsaSigner when: - def sig = signer.sign(10, result, up) + def sig = signer.sign(10, result, up.id) println("Signature: ${Hex.encodeHexString(sig.value)}") then: