StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added signature for native subscribe

Browse Source
This commit is contained in:
Maxksim Fomenkov
2022-09-14 10:14:34 +03:00
parent ea45462f7c
commit 08826ecbc4
27 changed files with 137 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt
View File

@@ -31,34 +31,40 @@ class BlockContainer(
json: ByteArray?,
val parsed: Any?,
val transactions: List<TxId> = 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<TransactionJson>()?.count() ?: 0 > 0
fun from(block: BlockJson<*>, raw: ByteArray, upstreamId: String): BlockContainer {
val hasTransactions = !block.transactions?.filterIsInstance<TransactionJson>().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)
}
}

2
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt
View File

@@ -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
}

2
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -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)
}

52
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt
View File

@@ -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<out Any> {
fun start(request: BlockchainOuterClass.NativeSubscribeRequest): Publisher<ResponseHolder> {
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
}
}

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt
View File

@@ -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}")

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt
View File

@@ -119,7 +119,7 @@ class EthereumDirectReader(
if (block == null) {
Mono.empty<BlockContainer>()
} else {
Mono.just(BlockContainer.from(block, blockbytes))
Mono.just(BlockContainer.from(block, blockbytes, "unknown"))
}
}
.doOnNext { block ->

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
View File

@@ -33,9 +33,10 @@ import java.util.concurrent.Executors
class EthereumRpcHead(
private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
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 =

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt
View File

@@ -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
}

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
View File

@@ -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)

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt
View File

@@ -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()
}
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt
View File

@@ -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")

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt
View File

@@ -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)
}
}

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt
View File

@@ -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)
}

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt
View File

@@ -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<Update> {
fun whenReplaced(prev: BlockContainer, source: String): Flux<Update> {
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 {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt
View File

@@ -94,7 +94,9 @@ class ProduceLogs(
txlog.topics,
txlog.transactionHash,
txlog.transactionIndex,
false
false,
update.upstreamId
)
}
oldMessages.put(LogReference(update.blockHash, update.transactionId), messages)

19
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceNewHeads.kt
View File

@@ -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<NewHeadMessage> {
return head.getFlux()
.map {
if (it.parsed != null) {
it.parsed as BlockJson<TransactionRefJson>
} 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<out TransactionRefJson> =
if (blockContainer.parsed != null) {
blockContainer.parsed as BlockJson<TransactionRefJson>
} else {
objectMapper.readValue(blockContainer.json, BlockJson::class.java)
}
}

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/HasUpstream.kt Normal file
View File

@@ -0,0 +1,5 @@
package io.emeraldpay.dshackle.upstream.ethereum.subscribe.json
interface HasUpstream {
val upstreamId: String
}

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/LogMessage.kt
View File

@@ -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

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadMessage.kt
View File

@@ -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

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
View File

@@ -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 ->

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
View File

@@ -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 ->

10
src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/EcdsaSigner.kt
View File

@@ -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()

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/NoSigner.kt
View File

@@ -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
}
}

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/ResponseSigner.kt
View File

@@ -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,

4
src/test/groovy/io/emeraldpay/dshackle/upstream/signature/EcdsaSignerSpec.groovy
View File

@@ -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: