diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 59387241a4..6bb76cdeca 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -21,7 +21,7 @@ jobs: strategy: matrix: os: [ubuntu-latest] - scala: [2.13.16, 2.12.20, 2.11.12] + scala: [2.13.18, 2.12.20, 2.11.12] java: [adopt@1.8] runs-on: ${{ matrix.os }} steps: @@ -64,7 +64,7 @@ jobs: strategy: matrix: os: [ubuntu-latest] - scala: [2.13.16, 2.12.20, 2.11.12] + scala: [2.13.18, 2.12.20, 2.11.12] java: [adopt@1.8] runs-on: ${{ matrix.os }} steps: diff --git a/.gitignore b/.gitignore index aad6dae921..eb475981a5 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,7 @@ devnet .ensime .ensime_cache/ scorex.yaml +*.iml # LLM reports on code analysis etc llm_generated diff --git a/avldb/build.sbt b/avldb/build.sbt index 96cf9327ee..c41c4d4321 100644 --- a/avldb/build.sbt +++ b/avldb/build.sbt @@ -2,7 +2,7 @@ import sbt.Keys.testFrameworks val scala211 = "2.11.12" val scala212 = "2.12.20" -val scala213 = "2.13.16" +val scala213 = "2.13.18" name := "avldb" @@ -30,7 +30,7 @@ val Versions = new { libraryDependencies ++= Seq( "ch.qos.logback" % "logback-classic" % "1.2.13", "com.google.guava" % "guava" % "23.0", - "org.scorexfoundation" %% "scrypto" % "2.3.0", + "org.scorexfoundation" %% "scrypto" % "3.1.1", "org.scalatest" %% "scalatest" % "3.2.19" % Test, "org.scalacheck" %% "scalacheck" % Versions.scalacheck(scalaVersion.value) % Test, Versions.scalatestplus(scalaVersion.value), diff --git a/build.sbt b/build.sbt index 14ee2cac11..be61e3d1a6 100644 --- a/build.sbt +++ b/build.sbt @@ -6,7 +6,7 @@ logLevel := Level.Debug // this values should be in sync with ergo-wallet/build.sbt val scala211 = "2.11.12" val scala212 = "2.12.20" -val scala213 = "2.13.16" +val scala213 = "2.13.18" lazy val commonSettings = Seq( organization := "org.ergoplatform", @@ -39,11 +39,11 @@ lazy val commonSettings = Seq( publishArtifact in (Compile, packageDoc) := false -val circeVersion = "0.13.0" +val circeVersion = "0.14.15" val akkaVersion = "2.6.10" val akkaHttpVersion = "10.2.4" -val sigmaStateVersion = "6.0.3" +val sigmaStateVersion = "6.0.6" val ficusVersion = "1.4.7" // for testing current sigmastate build (see sigmastate-ergo-it jenkins job) @@ -336,7 +336,7 @@ lazy val ergo = (project in file(".")) "com.github.scopt" %% "scopt" % "4.1.0", // API dependencies - "de.heikoseeberger" %% "akka-http-circe" % "1.20.0", + "de.heikoseeberger" %% "akka-http-circe" % "1.39.2", // app dependencies // jaxb-api is included only to avoid a runtime exception diff --git a/ergo-core/build.sbt b/ergo-core/build.sbt index 3021144c9a..45c19eb120 100644 --- a/ergo-core/build.sbt +++ b/ergo-core/build.sbt @@ -1,16 +1,16 @@ // this values should be in sync with root (i.e. ../build.sbt) val scala211 = "2.11.12" val scala212 = "2.12.20" -val scala213 = "2.13.16" +val scala213 = "2.13.18" val deps211 = Seq( "io.circe" %% "circe-core" % "0.10.0", "io.circe" %% "circe-generic" % "0.10.0", "io.circe" %% "circe-parser" % "0.10.0") val deps212 = Seq( - "io.circe" %% "circe-core" % "0.13.0", - "io.circe" %% "circe-generic" % "0.13.0", - "io.circe" %% "circe-parser" % "0.13.0") + "io.circe" %% "circe-core" % "0.14.15", + "io.circe" %% "circe-generic" % "0.14.15", + "io.circe" %% "circe-parser" % "0.14.15") publishMavenStyle := true Test / publishArtifact := false diff --git a/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/extension/ExtensionCandidate.scala b/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/extension/ExtensionCandidate.scala index 61513e3360..1bbfffc0f2 100644 --- a/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/extension/ExtensionCandidate.scala +++ b/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/extension/ExtensionCandidate.scala @@ -7,7 +7,6 @@ import scorex.crypto.authds.merkle.{BatchMerkleProof, Leaf, MerkleProof, MerkleT import scorex.crypto.hash.Digest32 import scorex.util.ModifierId import scala.annotation.nowarn -import scala.collection.mutable /** * Extension block section with header id not provided * @@ -49,9 +48,8 @@ class ExtensionCandidate(val fields: Seq[(Array[Byte], Array[Byte])]) { val indices = keys.flatMap(key => fields.find(_._1 sameElements key) .map(Extension.kvToLeaf) .map(kv => Leaf[Digest32](LeafData @@ kv)(Algos.hash).hash) - .flatMap(leafData => interlinksMerkleTree.elementsHashIndex.get( - new mutable.WrappedArray.ofByte(leafData)))) - if (indices.isEmpty) None else interlinksMerkleTree.proofByIndices(indices)(Algos.hash) + .flatMap(leafData => interlinksMerkleTree.indexByElementHash(leafData))) + if (indices.isEmpty) None else interlinksMerkleTree.proofByIndices(indices) } } diff --git a/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/popow/NipopowAlgos.scala b/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/popow/NipopowAlgos.scala index a441cfe9ff..0d76a9815e 100644 --- a/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/popow/NipopowAlgos.scala +++ b/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/popow/NipopowAlgos.scala @@ -75,6 +75,8 @@ class NipopowAlgos(val chainSettings: ChainSettings) { Int.MaxValue } + def hasValidPow(header: Header): Boolean = powScheme.validate(header).isSuccess + /** * Computes best score of a given chain. * The score value depends on number of ยต-superblocks in the given chain. @@ -96,6 +98,8 @@ class NipopowAlgos(val chainSettings: ChainSettings) { * end function */ def bestArg(chain: Seq[Header])(m: Int): Int = { + require(m >= 1, s"$m < 1") + @scala.annotation.tailrec def loop(level: Int, acc: Seq[(Int, Int)] = Seq.empty): Seq[(Int, Int)] = if (level == 0) { @@ -130,7 +134,6 @@ class NipopowAlgos(val chainSettings: ChainSettings) { val k = params.k val m = params.m - require(params.k >= 1, s"$k < 1") require(chain.lengthCompare(k + m) >= 0, s"Can not prove chain of size < ${k + m}") require(chain.head.header.isGenesis, "Can not prove non-anchored chain") diff --git a/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/popow/NipopowProof.scala b/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/popow/NipopowProof.scala index c963877ada..b922aca7c0 100644 --- a/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/popow/NipopowProof.scala +++ b/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/popow/NipopowProof.scala @@ -72,7 +72,12 @@ case class NipopowProof(popowAlgos: NipopowAlgos, * @return true if the proof is valid */ lazy val isValid: Boolean = { - this.hasValidConnections && this.hasValidHeights && this.hasValidProofs && this.hasValidDifficultyHeaders + PoPowParams.isValid(m, k) && + this.hasValidConnections && + this.hasValidHeights && + this.hasValidProofs && + this.hasValidDifficultyHeaders && + this.hasValidPow } /** @@ -155,6 +160,8 @@ case class NipopowProof(popowAlgos: NipopowAlgos, suffixHead.checkInterlinksProof() } + lazy val hasValidPow: Boolean = headersChain.forall(popowAlgos.hasValidPow) + } object NipopowProof { diff --git a/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/popow/PoPowParams.scala b/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/popow/PoPowParams.scala index 93aacc7339..151b4f4fe4 100644 --- a/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/popow/PoPowParams.scala +++ b/ergo-core/src/main/scala/org/ergoplatform/modifiers/history/popow/PoPowParams.scala @@ -1,5 +1,7 @@ package org.ergoplatform.modifiers.history.popow +import scala.util.Try + /** * NiPoPoW proof params from the KMZ17 paper * @@ -10,7 +12,20 @@ package org.ergoplatform.modifiers.history.popow * to calculate difficulty to be added to the proof). One-shot use means using the proof to just * to prove that a best chain contains some header (e.g. to work with a transaction corresponding * to the block header) + * @param minChainLength - minimal length of a proof's header chain acceptable for the params, always m + k. + * Kept as groundwork for the proof-length checks of the NiPoPoW parsing rework (#2461), + * not read by validation yet * */ -case class PoPowParams(m: Int, k: Int, continuous: Boolean) +final class PoPowParams private (val m: Int, val k: Int, val continuous: Boolean, val minChainLength: Int) + +object PoPowParams { + def isValid(m: Int, k: Int): Boolean = + m >= 1 && k >= 1 && m.toLong + k.toLong <= Int.MaxValue + + def apply(m: Int, k: Int, continuous: Boolean): Try[PoPowParams] = Try { + require(isValid(m, k), s"Invalid NiPoPoW parameters: m=$m, k=$k") + new PoPowParams(m, k, continuous, m + k) + } +} diff --git a/ergo-core/src/test/scala/org/ergoplatform/serialization/ContextExtensionSpec.scala b/ergo-core/src/test/scala/org/ergoplatform/serialization/ContextExtensionSpec.scala new file mode 100644 index 0000000000..ec8254ad1b --- /dev/null +++ b/ergo-core/src/test/scala/org/ergoplatform/serialization/ContextExtensionSpec.scala @@ -0,0 +1,79 @@ +package org.ergoplatform.serialization + +import org.ergoplatform.modifiers.mempool.{ErgoTransaction, ErgoTransactionSerializer} +import org.ergoplatform.settings.Constants.TrueTree +import org.ergoplatform.utils.ErgoCorePropertyTest +import org.ergoplatform.{ErgoBoxCandidate, Input} +import scorex.crypto.authds.ADKey +import scorex.util.encode.Base16 +import sigma.ast.IntConstant +import sigma.interpreter.{ContextExtension, ProverResult} + +import scala.util.Try + +/** + * Pins the sigma-state 6.0.6 change "reject negative-id vars in ContextExtension + * deserializer" and documents its consensus impact. + * + * The wire format itself did not change: var ids are written as one raw byte via + * signed `put(id)` (verified against the published sources of both 6.0.3 and 6.0.6). + * What changed is the reader only: 6.0.6 adds an explicit + * "Negative id of context extension variable" guard, so transactions carrying such + * extensions now fail at parse time with a `SerializerException`. + * + * Before 6.0.6 the same transaction could still be serialized/parsed successfully; + * negative ids were rejected later, at proving/verification time, with different + * exceptions (e.g., `NegativeArraySizeException`, `ArrayIndexOutOfBoundsException`). + * Those paths are covered by `ErgoTransactionSpec`. This spec adds the missing + * serializer-level and transaction-parser-level coverage that directly answers the + * review's confirm-item. + * + * Consensus impact: a hand-crafted transaction with a negative extension id could, + * in principle, be mined by an un-upgraded node and rejected by upgraded nodes. + * All known wallets/SDKs use small non-negative var ids, so no such transactions + * are known; the change is a soft-fork-style tightening that is safe once the + * majority of hashrate is upgraded. + */ +class ContextExtensionSpec extends ErgoCorePropertyTest { + + private val serializer = ContextExtension.serializer + + property("ContextExtension deserialization rejects negative var id") { + // wire layout: [values count][id][serialized value] + val bytes = serializer.toBytes(ContextExtension(Map(56.toByte -> IntConstant(0)))) + bytes(1) = 0xC8.toByte // id 56 -> -56 as signed byte + val parsed = Try(serializer.fromBytes(bytes)) + parsed.isFailure shouldBe true + parsed.failed.get.getMessage.contains("Negative id") shouldBe true + } + + property("ContextExtension serialization writes var id byte raw") { + // pins that the rejected wire bytes are producible by ordinary node code; + // the writer does not range-check the id + val bytes = serializer.toBytes(ContextExtension(Map((-56).toByte -> IntConstant(0)))) + bytes(1) shouldBe 0xC8.toByte + } + + property("ContextExtension valid ids round-trip") { + Seq(0.toByte, 127.toByte).foreach { id => + val ce = ContextExtension(Map(id -> IntConstant(1))) + serializer.fromBytes(serializer.toBytes(ce)) shouldBe ce + } + } + + property("ErgoTransaction parsing rejects negative context extension var id") { + val boxId = ADKey @@ Base16.decode("c95c2ccf55e03cac6659f71ca4df832d28e2375569cec178dcb17f3e2e5f7742").get + val input = Input( + boxId, + ProverResult(Array.emptyByteArray, ContextExtension(Map((-10).toByte -> IntConstant(0)))) + ) + val output = new ErgoBoxCandidate(1000000000L, TrueTree, 0) + val tx = ErgoTransaction(IndexedSeq(input), IndexedSeq.empty, IndexedSeq(output)) + + val bytes = ErgoTransactionSerializer.toBytes(tx) + val parsed = Try(ErgoTransactionSerializer.parseBytes(bytes)) + parsed.isFailure shouldBe true + parsed.failed.get.getMessage.contains("Negative id") shouldBe true + } + +} diff --git a/ergo-wallet/build.sbt b/ergo-wallet/build.sbt index af2bbb0097..43de051550 100644 --- a/ergo-wallet/build.sbt +++ b/ergo-wallet/build.sbt @@ -1,16 +1,16 @@ // this values should be in sync with root (i.e. ../build.sbt) val scala211 = "2.11.12" val scala212 = "2.12.20" -val scala213 = "2.13.16" +val scala213 = "2.13.18" val deps211 = Seq( "io.circe" %% "circe-core" % "0.10.0", "io.circe" %% "circe-generic" % "0.10.0", "io.circe" %% "circe-parser" % "0.10.0") val deps212 = Seq( - "io.circe" %% "circe-core" % "0.13.0", - "io.circe" %% "circe-generic" % "0.13.0", - "io.circe" %% "circe-parser" % "0.13.0") + "io.circe" %% "circe-core" % "0.14.15", + "io.circe" %% "circe-generic" % "0.14.15", + "io.circe" %% "circe-parser" % "0.14.15") libraryDependencies ++= Seq( "org.scodec" %% "scodec-bits" % "1.1.34", diff --git a/src/it/resources/parameters-template.txt b/src/it/resources/parameters-template.txt deleted file mode 100644 index 17473bdc9c..0000000000 --- a/src/it/resources/parameters-template.txt +++ /dev/null @@ -1,198 +0,0 @@ -paths: - - /blocks: - post: - - | - { - "header": { - "votes": "000000", - "difficulty": "291", - "timestamp": 1538572701768, - "size": 607, - "stateRoot": "c2d3cfb7482c9edc4b1214b830032b93556af6a4a9224c7154cbf185bc00c15316", - "height": 32693, - "nBits": 33628928, - "version": 1, - "id": "00d2a8d21113598ea924329f9520905693e914bac6235255b74fd3b8016171aa", - "adProofsRoot": "a84f62a669fb3684308ea609af6cd831b939b70210307e143a05321ce8efeda2", - "transactionsRoot": "cc32add0ada11b6a81f07b61a6c606b2277af9999b453538d32b3c409630bce1", - "extensionHash": "0e5751c026e543b2e8ab2eb06099daa1d1e5df47778f7787faab45cdf12fe3a8", - "powSolutions" : { - "pk" : "0350e25cee8562697d55275c96bb01b34228f9bd68fd9933f2a25ff195526864f5", - "w" : "032e3f5edb88f3cc7384bfedc892fc8dcb10d7a3bf3741d08a62cc701848d0932c", - "n" : "0000000000000000", - "d" : 603083798111021851164432213586916186738093029948325633833495911523787854249 - }, - "parentId": "002390f165396f855f53b928e469ba89a2107784479423a2db66b3acccef78e9" - }, - "blockTransactions": { - "headerId": "00d2a8d21113598ea924329f9520905693e914bac6235255b74fd3b8016171aa", - "transactions": [ - { - "id": "20687df938ee4b3c50b7e145f84bb881f9518465c0877bf77a681241fe1a60e6", - "inputs": [ - { - "boxId": "0ab3aa13f4a82cfb4e8031a08f0ebdf2e969350deb897a171ec888bb8a6cc2a0", - "spendingProof": { - "proofBytes": "e8f50a65ee577e55b6ef100bfe97597e717b6792c94c749407d37547a31a9bba133be88e0a634bb479d3caffe9567059816cb8780a3bb373", - "extension": {} - } - }, - { - "boxId": "2a14b5969a287413918e9af76f939b47c6395ae84c65fa040f11e769e754c474", - "spendingProof": { - "proofBytes": "4b40eb0ca07bff34b4c07cc8c6a3eabc4f7634641e3941670734f59d68a2a946cf1e2a222f08a642c6b5228d331d9ab3120f10d1958a5659", - "extension": {} - } - } - ], - "dataInputs" : [], - "outputs": [ - { - "boxId": "08667225fa6beb560627ba02b5389929db756add5b65bda8aea7920088b2cc8c", - "value": 100000, - "ergoTree": "100101017300", - "assets": [], - "additionalRegisters": {}, - "creationHeight": 1149 - }, - { - "boxId": "a27ff81f905bfec9e4a551d95322a0c53259e44cd263cf72fa7d1eabc715fefe", - "value": 900000000, - "ergoTree": "1001070361a3df05f414e9b01394487c4cc9838857575408677ddc166047a52f93e4ed26cd7300", - "assets": [], - "additionalRegisters": {}, - "creationHeight": 1149 - } - ], - "size": 363 - }, - { - "id": "e0b057427cbe654bc872631e06bb0d96f91eb5a9779f4b4b1503c377582ca636", - "inputs": [ - { - "boxId": "aaacb6f598587059b770289abcbae0bc3dd949f527052f2ff82230f8411e67b1", - "spendingProof": { - "proofBytes": "28a6e09627092c63151fc10025abe30f368153e4255f26b90768bf60023d321ebde22692f2b412a7e964030333f615dc783b001bcb4865f7", - "extension": {} - } - } - ], - "dataInputs" : [ - { - "boxId" : "fec8fa6499dc086098d1589fdeefdd195def862098b84c9a7d85aaab28e92d0e" - }, - { - "boxId" : "fecf6c5508e87f7e71170f4fc88053539b7b70831a72b42090d7d480ae69e52f" - }, - { - "boxId" : "fee3087d7ecd4fbef2bea80dae2d0ef86e690c26a68084123c9fa1036761e82b" - } - ], - "outputs": [ - { - "boxId": "74c503605218f6e382710bc0cb6716a63e887a2a10252cdbb4662d49d660069c", - "value": 100000, - "ergoTree": "100101017300", - "assets": [], - "additionalRegisters": {}, - "creationHeight": 1149 - }, - { - "boxId": "cf5a2dbc79d3152305804e7109743d2ed538a5ee73e7fe7f366213c4d028f683", - "value": 500000000, - "ergoTree": "1001070361a3df05f414e9b01394487c4cc9838857575408677ddc166047a52f93e4ed26cd7300", - "assets": [], - "additionalRegisters": {}, - "creationHeight": 1149 - } - ], - "size": 184 - } - ], - "size": 4054 - }, - "extension": { - "headerId": "00d2a8d21113598ea924329f9520905693e914bac6235255b74fd3b8016171aa", - "digest": "0e5751c026e543b2e8ab2eb06099daa1d1e5df47778f7787faab45cdf12fe3a8", - "fields": [] - }, - "adProofs": { - "headerId": "00d2a8d21113598ea924329f9520905693e914bac6235255b74fd3b8016171aa", - "proofBytes": "f7ab7f72f0bd0a761308dc0ebe5d9011745fd2aea137918208987aae87a9169cc5f7ff488e5333b6c535d2ce0a03cfbb8265d95a93323323", - "digest": "a84f62a669fb3684308ea609af6cd831b939b70210307e143a05321ce8efeda2", - "size": 28540 - }, - "size": 33201 - } - - /blocks/lastHeaders/{count}: - path_parameters: - - count: @lastHeadersCount - - /blocks/at/{blockHeight}: - path_parameters: - - blockHeight: @blockHeight - - /blocks/{headerId}: - path_parameters: - - headerId: @headerId - - /blocks/{headerId}/header: - path_parameters: - - headerId: @headerId - - /blocks/{headerId}/transactions: - path_parameters: - - headerId: @headerId - - /transactions: - post: - - | - { - "id": "2ab9da11fc216660e974842cc3b7705e62ebb9e0bf5ff78e53f9cd40abadd117", - "inputs": [ - { - "boxId": "1ab9da11fc216660e974842cc3b7705e62ebb9e0bf5ff78e53f9cd40abadd117", - "spendingProof": { - "proofBytes": "46dc83d572290479218cfe2a8cb9a0d979de4a61da6a96260a53479683f55a2b561ba4ec1faea4a0d4d19efc960a188ebe3fb8fb83e39d3b", - "extension": { - "1": "0101" - } - } - } - ], - "dataInputs" : [ - { - "boxId" : "fee3087d7ecd4fbef2bea80dae2d0ef86e690c26a68084123c9fa1036761e82b" - } - ], - "outputs": [ - { - "boxId": "1ab9da11fc216660e974842cc3b7705e62ebb9e0bf5ff78e53f9cd40abadd117", - "value": 147, - "ergoTree": "1001070361a3df05f414e9b01394487c4cc9838857575408677ddc166047a52f93e4ed26cd7300", - "assets": [ - { - "tokenId": "4ab9da11fc216660e974842cc3b7705e62ebb9e0bf5ff78e53f9cd40abadd117", - "amount": 1000 - } - ], - "additionalRegisters": { - "R4": "0101" - }, - "creationHeight": 1149 - } - ], - "size": 0 - } - - /utils/hash/blake2b: - post: - - '"123qwe"' - - '""' - - '"aaaaaaaaaa"' - - /utils/seed/{length}: - path_parameters: - - length: 123 \ No newline at end of file diff --git a/src/it/scala/org/ergoplatform/it/OpenApiSpec.scala b/src/it/scala/org/ergoplatform/it/OpenApiSpec.scala deleted file mode 100644 index 68653494c4..0000000000 --- a/src/it/scala/org/ergoplatform/it/OpenApiSpec.scala +++ /dev/null @@ -1,75 +0,0 @@ -package org.ergoplatform.it - -import java.io.{File, PrintWriter} - -import com.typesafe.config.Config -import org.ergoplatform.it.container.{ - ApiChecker, - ApiCheckerConfig, - IntegrationSuite, - Node -} -import org.scalatest.flatspec.AnyFlatSpec - -import scala.concurrent.duration._ -import scala.concurrent.{Await, Future} -import scala.io.Source - -class OpenApiSpec extends AnyFlatSpec with IntegrationSuite { - - val expectedHeight: Int = 2 - val paramsFilePath: String = "/tmp/parameters.yaml" - val paramsTemplatePath: String = "src/it/resources/parameters-template.txt" - - val offlineGeneratingPeer: Config = offlineGeneratingPeerConfig - .withFallback(nodeSeedConfigs.head) - .withFallback(allowLocalConfig) - - // `lazy` so the container is only started when a test actually touches `node`. - // The single test below is currently `ignore`d (the openapi-checker image is gone), - // so without `lazy` we would start and tear down a node for nothing. - lazy val node: Node = docker.startDevNetNode(offlineGeneratingPeer).get - - def renderTemplate(template: String, varMapping: Map[String, String]): String = - varMapping - .foldLeft(template) { case (s, (k, v)) => s.replaceAll(s"@$k", v) } - - def createParamsFile(params: Map[String, String]): Unit = { - val template: String = - Source.fromFile(paramsTemplatePath).getLines.map(_ + "\n").mkString - val writer: PrintWriter = new PrintWriter(new File(paramsFilePath)) - writer.write(renderTemplate(template, params)) - writer.close() - } - - it should "OpenApi specification check" ignore { - val result: Future[Unit] = node - .waitForHeight(expectedHeight) - .flatMap { _ => - node.headerIdsByHeight(expectedHeight) - } - .map { headerIds => - createParamsFile( - Map( - "blockHeight" -> expectedHeight.toString, - "lastHeadersCount" -> expectedHeight.toString, - "headerId" -> headerIds.head - ) - ) - - val apiAddressToCheck: String = - s"${node.nodeInfo.networkIpAddress}:${node.nodeInfo.containerApiPort}" - val specFilePath: String = - new File("src/main/resources/api/openapi.yaml").getAbsolutePath - val checker: ApiChecker = docker - .startOpenApiChecker( - ApiCheckerConfig(apiAddressToCheck, specFilePath, paramsFilePath) - ) - .get - - docker.waitContainer(checker.containerId).awaitStatusCode() shouldBe 0 - } - - Await.result(result, 2.minutes) - } -} diff --git a/src/it/scala/org/ergoplatform/it/container/ApiChecker.scala b/src/it/scala/org/ergoplatform/it/container/ApiChecker.scala deleted file mode 100644 index 6c9d8bfe32..0000000000 --- a/src/it/scala/org/ergoplatform/it/container/ApiChecker.scala +++ /dev/null @@ -1,5 +0,0 @@ -package org.ergoplatform.it.container - -case class ApiCheckerConfig(apiAddressToCheck: String, specFilePath: String, paramsFilePath: String) - -case class ApiChecker(containerId: String, config: ApiCheckerConfig) diff --git a/src/it/scala/org/ergoplatform/it/container/Docker.scala b/src/it/scala/org/ergoplatform/it/container/Docker.scala index c392420e6d..4c9e5f0d2f 100644 --- a/src/it/scala/org/ergoplatform/it/container/Docker.scala +++ b/src/it/scala/org/ergoplatform/it/container/Docker.scala @@ -67,7 +67,6 @@ class Docker( private val client: DockerClient = DockerClientImpl.getInstance(configStandart, httpDockerClient) private var nodeRepository = Seq.empty[Node] - private var apiCheckerOpt: Option[ApiChecker] = None private val isStopped = new AtomicBoolean(false) // This should be called after client is ready but before network created. @@ -120,19 +119,6 @@ class Docker( def waitContainer(id: String): WaitContainerResultCallback = client.waitContainerCmd(id).start() - def startOpenApiChecker(checkerInfo: ApiCheckerConfig): Try[ApiChecker] = Try { - val ip: String = ipForNode(999, networkSeed) - val containerId: String = buildApiCheckerContainerCmd(checkerInfo, ip).exec().getId - connectToNetwork(containerId, ip) - client.startContainerCmd(containerId).exec() - - log.info(s"Started ApiChecker: $containerId") - - val checker: ApiChecker = ApiChecker(containerId, checkerInfo) - apiCheckerOpt = Some(checker) - checker - } - private def startNode( networkType: NetworkType, nodeSpecificConfig: Config, @@ -236,30 +222,6 @@ class Docker( actualConfig } - private def buildApiCheckerContainerCmd( - checkerInfo: ApiCheckerConfig, - ip: String - ): CreateContainerCmd = { - val hostConfig: HostConfig = new HostConfig() - .withBinds( - new Bind(checkerInfo.specFilePath, new Volume("/opt/ergo/openapi.yaml")), - new Bind(checkerInfo.paramsFilePath, new Volume("/opt/ergo/parameters.yaml")) - ) - - client - .createContainerCmd(ApiCheckerImageStable) - .withCmd( - "openapi.yaml", - "--api", - s"http://${checkerInfo.apiAddressToCheck}", - "--parameters", - "parameters.yaml" - ) - .withHostConfig(hostConfig) - .withHostName(networkName) - .withIpv4Address(ip) - } - private def buildPeerContainerCmd( networkType: NetworkType, nodeConfig: Config, @@ -456,11 +418,6 @@ class Docker( saveNodeLogs() - apiCheckerOpt.foreach { checker => - saveLogs(checker.containerId, "openapi-checker") - client.removeContainerCmd(checker.containerId).withForce(true).exec() - } - nodeRepository foreach { node => client.removeContainerCmd(node.containerId).withForce(true).exec() } @@ -565,9 +522,7 @@ class Docker( object Docker extends IntegrationTestConstants { - val ErgoImageLatest: String = "org.ergoplatform/ergo" - val ApiCheckerImageLatest: String = "andyceo/openapi-checker" - val ApiCheckerImageStable: String = "andyceo/openapi-checker:0.1.0-openapi-core-0.5.0" // not present in docker anymore + val ErgoImageLatest: String = "org.ergoplatform/ergo" val dockerImageLabel = "ergo-integration-tests" val networkNamePrefix: String = "ergo-itest-" diff --git a/src/main/resources/api/openapi.yaml b/src/main/resources/api/openapi.yaml index 30ec41d47e..c35ea09884 100644 --- a/src/main/resources/api/openapi.yaml +++ b/src/main/resources/api/openapi.yaml @@ -1,7 +1,7 @@ openapi: "3.0.2" info: - version: "6.0.4" + version: "6.0.5" title: Ergo Node API description: API docs for Ergo Node. Models are shared between all Ergo products contact: diff --git a/src/main/resources/application.conf b/src/main/resources/application.conf index 4bbea9b00a..bc8db13f0e 100644 --- a/src/main/resources/application.conf +++ b/src/main/resources/application.conf @@ -446,7 +446,7 @@ scorex { nodeName = "ergo-node" # Network protocol version to be sent in handshakes - appVersion = 6.0.4 + appVersion = 6.0.5 # Network agent name. May contain information about client code # stack, starting from core code-base up to the end graphical interface. diff --git a/src/main/resources/logback.xml b/src/main/resources/logback.xml index 2929ac6ade..d2f0932a66 100644 --- a/src/main/resources/logback.xml +++ b/src/main/resources/logback.xml @@ -30,6 +30,10 @@ + + + diff --git a/src/main/scala/org/ergoplatform/http/ErgoHttpService.scala b/src/main/scala/org/ergoplatform/http/ErgoHttpService.scala index c331e005e2..eb99d040a2 100644 --- a/src/main/scala/org/ergoplatform/http/ErgoHttpService.scala +++ b/src/main/scala/org/ergoplatform/http/ErgoHttpService.scala @@ -8,6 +8,7 @@ import akka.http.scaladsl.server.Directive0 import akka.http.scaladsl.server.directives.RouteDirectives import scorex.core.api.http.{ApiErrorHandler, ApiRejectionHandler, ApiRoute, CorsHandler} import akka.http.scaladsl.model.headers._ +import scorex.util.ScorexLogging import scala.collection.immutable @@ -15,7 +16,7 @@ final case class ErgoHttpService( apiRoutes: Seq[ApiRoute], swaggerRoute: SwaggerRoute, panelRoute: NodePanelRoute -)(implicit val system: ActorSystem) extends CorsHandler { +)(implicit val system: ActorSystem) extends CorsHandler with ScorexLogging { def rejectionHandler: RejectionHandler = ApiRejectionHandler.rejectionHandler @@ -36,15 +37,41 @@ final case class ErgoHttpService( super.respondWithHeaders(corsResponseHeaders) } + /** + * Logs every query served by the node's HTTP interface: method, relative URI (path and query + * string), response status and how long it took. + * + * Bodies are deliberately not logged, as requests carry secrets (a mnemonic on + * `/wallet/restore`, a password on `/wallet/unlock`, and so on) and responses can be large. + * + * Off by default, since the root logger is at INFO. To switch it on, add to `logback.xml`: + * {{{ + * + * }}} + * When it is off, the message is never built: `log.debug` is a macro guarded by `isDebugEnabled`. + */ + private val logQueries: Directive0 = + extractRequest.flatMap { request => + val startTime = System.currentTimeMillis() + mapResponse { response => + val elapsedMs = System.currentTimeMillis() - startTime + log.debug(s"${request.method.value} ${request.uri.toRelative} - " + + s"${response.status.intValue()} in $elapsedMs ms") + response + } + } + val compositeRoute: Route = - handleRejections(rejectionHandler) { - handleExceptions(exceptionHandler) { - corsHandler { - apiR ~ - apiSpecR ~ - swaggerRoute.route ~ - panelRoute.route ~ - redirectToSwaggerR + logQueries { + handleRejections(rejectionHandler) { + handleExceptions(exceptionHandler) { + corsHandler { + apiR ~ + apiSpecR ~ + swaggerRoute.route ~ + panelRoute.route ~ + redirectToSwaggerR + } } } } diff --git a/src/main/scala/org/ergoplatform/http/api/ErgoPeersApiRoute.scala b/src/main/scala/org/ergoplatform/http/api/ErgoPeersApiRoute.scala index 88bf56eeea..6fa0b79f80 100644 --- a/src/main/scala/org/ergoplatform/http/api/ErgoPeersApiRoute.scala +++ b/src/main/scala/org/ergoplatform/http/api/ErgoPeersApiRoute.scala @@ -42,10 +42,12 @@ class ErgoPeersApiRoute(peerManager: ActorRef, } def allPeers: Route = (path("all") & get) { - val result = askActor[Map[InetSocketAddress, PeerInfo]](peerManager, GetAllPeers).map { - _.map { case (address, peerInfo) => - PeerInfoResponse.fromAddressAndInfo(address, peerInfo) - } + val result = askActor[Map[InetSocketAddress, PeerInfo]](peerManager, GetAllPeers).map { peers => + peers.toSeq + .sortBy(_._1.toString) + .map { case (address, peerInfo) => + PeerInfoResponse.fromAddressAndInfo(address, peerInfo) + } } ApiResponse(result) } diff --git a/src/main/scala/org/ergoplatform/modifiers/history/popow/NipopowProverWithDbAlgs.scala b/src/main/scala/org/ergoplatform/modifiers/history/popow/NipopowProverWithDbAlgs.scala index d4bffa263d..94dc4924fa 100644 --- a/src/main/scala/org/ergoplatform/modifiers/history/popow/NipopowProverWithDbAlgs.scala +++ b/src/main/scala/org/ergoplatform/modifiers/history/popow/NipopowProverWithDbAlgs.scala @@ -29,7 +29,6 @@ object NipopowProverWithDbAlgs { val k = params.k val m = params.m - require(params.k >= 1, s"$k < 1") require(histReader.headersHeight >= k + m, s"Can not prove chain of size < ${k + m}") def linksWithIndexes(header: PoPowHeader): Seq[(ModifierId, Int)] = header.interlinks.tail.reverse.zipWithIndex diff --git a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala index ced63d1e8a..5a82b55918 100644 --- a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala +++ b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala @@ -1473,7 +1473,7 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, log.info(s"Penalize spamming peer $peer for too costly transaction $id") penalizeSpammingPeer(peer) case _ => - log.info(s"Penalize peer $peer for too costly transaction $id (reason: $error)") + log.info(s"Penalize peer $peer for transaction $id (reason: $error)") penalizeMisbehavingPeer(peer) } } diff --git a/src/main/scala/org/ergoplatform/network/message/BasicMessagesRepo.scala b/src/main/scala/org/ergoplatform/network/message/BasicMessagesRepo.scala index b7b5842154..151d61099f 100644 --- a/src/main/scala/org/ergoplatform/network/message/BasicMessagesRepo.scala +++ b/src/main/scala/org/ergoplatform/network/message/BasicMessagesRepo.scala @@ -5,7 +5,6 @@ import org.ergoplatform.nodeView.state.SnapshotsInfo import org.ergoplatform.nodeView.state.UtxoState.{ManifestId, SubtreeId} import org.ergoplatform.network.message.MessageConstants.MessageCode import scorex.crypto.hash.Digest32 -import scorex.util.Extensions._ import scorex.util.serialization.{Reader, Writer} import org.ergoplatform.sdk.wallet.Constants.ModifierIdLength @@ -53,7 +52,7 @@ class PeersSpec(peersLimit: Int) extends MessageSpecV1[Seq[PeerSpec]] { } override def parse(r: Reader): Seq[PeerSpec] = { - val length = r.getUInt().toIntExact + val length = r.getUIntExact() require(length <= peersLimit, s"Too many peers. $length exceeds limit $peersLimit") (0 until length).map { _ => PeerSpecSerializer.parse(r) @@ -101,7 +100,7 @@ object SnapshotsInfoSpec extends MessageSpecV1[SnapshotsInfo] { override def parse(r: Reader): SnapshotsInfo = { require(r.remaining <= SizeLimit, s"Too big SnapshotsInfo message: ${r.remaining} bytes found, $SizeLimit max expected.") - val length = r.getUInt().toIntExact + val length = r.getUIntExact() val manifests = (0 until length).map { _ => val height = r.getInt() val manifest = Digest32 @@ r.getBytes(ModifierIdLength) @@ -151,7 +150,7 @@ object ManifestSpec extends MessageSpecV1[Array[Byte]] { override def parse(r: Reader): Array[Byte] = { require(r.remaining <= SizeLimit, s"Too big Manifest message.") - val length = r.getUInt().toIntExact + val length = r.getUIntExact() r.getBytes(length) } @@ -196,7 +195,7 @@ object UtxoSnapshotChunkSpec extends MessageSpecV1[Array[Byte]] { override def parse(r: Reader): Array[Byte] = { require(r.remaining <= SizeLimit, s"Too big UtxoSnapshotChunk message.") - val length = r.getUInt().toIntExact + val length = r.getUIntExact() r.getBytes(length) } diff --git a/src/main/scala/org/ergoplatform/network/peer/PeerDatabase.scala b/src/main/scala/org/ergoplatform/network/peer/PeerDatabase.scala index 8a3c624791..c3295c1582 100644 --- a/src/main/scala/org/ergoplatform/network/peer/PeerDatabase.scala +++ b/src/main/scala/org/ergoplatform/network/peer/PeerDatabase.scala @@ -7,20 +7,32 @@ import java.io.{ ObjectOutputStream } import java.net.{InetAddress, InetSocketAddress} +import java.util.concurrent.ThreadLocalRandom import org.ergoplatform.settings.ErgoSettings import scorex.db.LDBFactory import scorex.util.ScorexLogging +import scala.collection.mutable import scala.concurrent.duration._ import scala.util.{Failure, Success, Try} /** - * In-memory peer database implementation supporting temporal blacklisting. + * In-memory peer database with temporal blacklisting and peer count cap. */ -final class PeerDatabase(settings: ErgoSettings) extends ScorexLogging { +final class PeerDatabase( + settings: ErgoSettings, + private[peer] val maxKnownPeers: Int = PeerDatabase.MaxKnownPeers +) extends ScorexLogging { private val persistentStore = LDBFactory.createKvDb(s"${settings.directory}/peers") + private case class LoadedPeer( + lastHandshake: Long, + address: InetSocketAddress, + peerInfo: PeerInfo, + keyBytes: Array[Byte] + ) + /** * banned peer ip -> ban expiration timestamp */ @@ -62,30 +74,203 @@ final class PeerDatabase(settings: ErgoSettings) extends ScorexLogging { } /* - * Load peers from persistent storage + * Number of store keys removed per batch while loading peers, so that + * cleanup of a severely oversized database does not build one huge in-memory batch. + */ + private val RemovalBatchSize = 1024 + + /* + * Load peers from persistent storage. + * + * Retention is driven by recency, not store iteration order: while streaming over + * the store we keep at most `maxKnownPeers` entries with the newest `lastHandshake` + * values, using a bounded min-heap keyed by `lastHandshake`, so peak extra memory + * stays O(maxKnownPeers + RemovalBatchSize) and an oversized or malformed database + * cannot OOM the node on startup. + * + * Oversized, unparseable, duplicated (same address, older handshake) and excess + * entries are physically removed from the store in bounded batches during the scan. */ private def loadPeers: Try[Map[InetSocketAddress, PeerInfo]] = Try { - var peers = Map.empty[InetSocketAddress, PeerInfo] - for ((addr, peer) <- persistentStore.getAll) { - val address = deserialize(addr).asInstanceOf[InetSocketAddress] - val peerInfo = PeerInfoSerializer.parseBytes(peer) - peers += address -> peerInfo + val kept = mutable.HashMap.empty[InetSocketAddress, LoadedPeer] + // min-heap by lastHandshake (oldest at the head); may contain stale entries + // superseded by a newer record for the same address, cleaned lazily on eviction + val oldestFirst = + mutable.PriorityQueue.empty[LoadedPeer](Ordering.by[LoadedPeer, Long](_.lastHandshake).reverse) + val keysToRemove = mutable.ArrayBuffer.empty[Array[Byte]] + var removedRecords = 0L + + def flushRemovalBuffer(force: Boolean = false): Unit = { + if (keysToRemove.nonEmpty && (force || keysToRemove.length >= RemovalBatchSize)) { + flushKeysToRemove(keysToRemove.toArray) + removedRecords += keysToRemove.length + keysToRemove.clear() + } + } + + def dropKey(key: Array[Byte]): Unit = { + keysToRemove += key + flushRemovalBuffer() + } + + persistentStore.getAll.foreach { case (key, value) => + if (key.length > PeerDatabase.MaxSerializedPeerAddressSize || + value.length > PeerDatabase.MaxSerializedPeerInfoSize) { + log.warn( + s"Dropping oversized peer entry from database: key=${key.length} bytes, " + + s"value=${value.length} bytes" + ) + dropKey(key) + } else { + val addressTry = Try(deserialize(key).asInstanceOf[InetSocketAddress]) + val peerInfoTry = PeerInfoSerializer.parseBytesTry(value) + (addressTry, peerInfoTry) match { + case (Success(address), Success(peerInfo)) => + val loaded = LoadedPeer(peerInfo.lastHandshake, address, peerInfo, key) + kept.get(address) match { + case Some(existing) if existing.lastHandshake >= loaded.lastHandshake => + dropKey(key) + case Some(existing) => + kept(address) = loaded + oldestFirst.enqueue(loaded) + dropKey(existing.keyBytes) + case None if kept.size < maxKnownPeers => + kept(address) = loaded + oldestFirst.enqueue(loaded) + case None => + // evict the oldest kept peer if the loaded one is newer + while (oldestFirst.headOption.exists(p => kept.get(p.address).forall(_ != p))) { + oldestFirst.dequeue() + } + oldestFirst.headOption match { + case Some(oldest) if loaded.lastHandshake > oldest.lastHandshake => + kept -= oldest.address + kept(address) = loaded + oldestFirst.enqueue(loaded) + dropKey(oldest.keyBytes) + case _ => + dropKey(key) + } + } + case _ => + log.warn(s"Unable to deserialize peer entry from database, removing it") + dropKey(key) + } + } + } + flushRemovalBuffer(force = true) + + if (removedRecords > 0) { + log.info(s"Removed $removedRecords malformed, oversized or excess peer entries from database on startup") + } + kept.map { case (address, loaded) => address -> loaded.peerInfo }.toMap + } + + private def flushKeysToRemove(keys: Array[Array[Byte]]): Unit = { + if (keys.nonEmpty) { + persistentStore.remove(keys) match { + case Success(_) => // ok + case Failure(ex) => + log.warn("Unable to remove dropped peer entries from persistent store", ex) + } } - peers } def get(peer: InetSocketAddress): Option[PeerInfo] = peers.get(peer) - def addOrUpdateKnownPeer(peerInfo: PeerInfo): Unit = { + def addOrUpdateKnownPeer( + peerInfo: PeerInfo, + connectedPeers: Set[InetSocketAddress] = Set.empty + ): Unit = { if (!peerInfo.peerSpec.declaredAddress.exists(x => isBlacklisted(x.getAddress))) { peerInfo.peerSpec.address.foreach { address => - log.debug(s"Updating peer info for $address") - peers += address -> peerInfo - persistentStore.insert(serialize(address), PeerInfoSerializer.toBytes(peerInfo)) + if (peers.contains(address)) { + log.debug(s"Updating peer info for $address") + updatePeer(address, peerInfo) + } else if (peers.size < maxKnownPeers || + makeRoomForPeer(peerInfo.lastHandshake, connectedPeers)) { + log.debug(s"Adding peer info for $address") + updatePeer(address, peerInfo) + } else { + log.debug(s"Peer database is full, ignoring $address") + } } } } + private def updatePeer(address: InetSocketAddress, peerInfo: PeerInfo): Unit = { + peers += address -> peerInfo + persistentStore.insert(serialize(address), PeerInfoSerializer.toBytes(peerInfo)) + } + + /** + * Evict the oldest known peer (by lastHandshake) from a random sample to make room + * for a new peer, but never evict a currently connected peer. + * + * Note: a candidate with `lastHandshake == 0` (a peer we have not handshaked with yet) + * can never displace an existing peer. This is an intentional anti-spam policy: data + * about not-yet-verified peers must not evict verified ones. + * + * @param candidateHandshake - lastHandshake of the peer we want to insert + * @return true if room was made, false otherwise + */ + private def makeRoomForPeer( + candidateHandshake: Long, + connectedPeers: Set[InetSocketAddress] + ): Boolean = { + val EvictionSampleSize = 16 + val oldest = randomPeerSample(EvictionSampleSize) + .filterNot { case (address, _) => connectedPeers.contains(address) } + .sortBy(_._2.lastHandshake) + .headOption + + oldest match { + case Some((oldestAddress, oldestInfo)) + if candidateHandshake > oldestInfo.lastHandshake => + log.info( + s"Evicting peer $oldestAddress with lastHandshake " + + s"${oldestInfo.lastHandshake} to make room for a newer peer" + ) + remove(oldestAddress) + true + case _ => + false + } + } + + /** + * Select a small random slice of known peers to consider for eviction. + * The slice is contiguous in the map's iteration order and bounded by + * `sampleSize`, so the cost stays low even when the peer set is large. + */ + private def randomPeerSample(sampleSize: Int): Seq[(InetSocketAddress, PeerInfo)] = { + if (peers.isEmpty) { + Seq.empty + } else { + val sample = math.min(sampleSize, peers.size) + val start = ThreadLocalRandom.current().nextInt(peers.size - sample + 1) + peers.slice(start, start + sample).toSeq + } + } + + /** + * Remove peers whose lastHandshake is older than 60 days, excluding connected peers + * and peers without a successful handshake (`lastHandshake == 0`, e.g. discovered + * but not yet tried peers and unavailable configured seeds), so that untried peers + * are not purged shortly after being discovered. + */ + def removeOldPeers(connectedPeers: Set[InetSocketAddress] = Set.empty): Unit = { + val cutoff = System.currentTimeMillis() - PeerDatabase.KnownPeerMaxAgeMs + val toRemove = peers.collect { + case (address, info) + if !connectedPeers.contains(address) && + info.lastHandshake != 0 && + info.lastHandshake < cutoff => + address + } + toRemove.foreach(remove) + } + def addToBlacklist(socketAddress: InetSocketAddress, penaltyType: PenaltyType): Unit = { remove(socketAddress) Option(socketAddress.getAddress).foreach { address => @@ -100,7 +285,7 @@ final class PeerDatabase(settings: ErgoSettings) extends ScorexLogging { } } - def removeFromBlacklist(address: InetAddress): Unit = { + private def removeFromBlacklist(address: InetAddress): Unit = { log.info(s"$address removed from blacklist") blacklist -= address } @@ -112,6 +297,11 @@ final class PeerDatabase(settings: ErgoSettings) extends ScorexLogging { def knownPeers: Map[InetSocketAddress, PeerInfo] = peers + /** + * Close the underlying persistent store. + */ + def close(): Unit = persistentStore.close() + def blacklistedPeers: Seq[InetAddress] = blacklist.map { case (address, bannedTill) => @@ -188,3 +378,36 @@ final class PeerDatabase(settings: ErgoSettings) extends ScorexLogging { (360 * 10).days.toMillis } } + +object PeerDatabase { + + /** + * Hardcoded cap on the total number of known peers. + */ + val MaxKnownPeers: Int = 32768 + + /** + * Serialized peer info size must stay below this bound. The value is twice + * the maximum handshake size (8KB) to leave a comfortable margin while still + * preventing a single malformed/crafted entry from consuming a lot of memory. + */ + private[peer] val MaxSerializedPeerInfoSize: Int = 16384 + + /** + * Serialized peer address (InetSocketAddress Java serialization) size bound. + * Legitimate hostnames can be up to 253 characters, so leave plenty of headroom. + */ + private[peer] val MaxSerializedPeerAddressSize: Int = 1024 + + /** + * Hardcoded maximum age (60 days) for a known peer's lastHandshake. + * Peers with `lastHandshake == 0` (never handshaked) are exempt from age cleanup. + */ + val KnownPeerMaxAgeMs: Long = 60.days.toMillis + + /** + * Hardcoded interval (24 hours) between cleanup runs. + */ + val KnownPeerCleanupIntervalMs: Long = 24.hours.toMillis + +} diff --git a/src/main/scala/org/ergoplatform/network/peer/PeerManager.scala b/src/main/scala/org/ergoplatform/network/peer/PeerManager.scala index 96b7b392d7..680a1b1cca 100644 --- a/src/main/scala/org/ergoplatform/network/peer/PeerManager.scala +++ b/src/main/scala/org/ergoplatform/network/peer/PeerManager.scala @@ -1,7 +1,12 @@ package org.ergoplatform.network.peer import java.net.{InetAddress, InetSocketAddress} +import java.util.concurrent.ThreadLocalRandom import akka.actor.{Actor, ActorRef, ActorSystem, Props} +import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages.{ + DisconnectedPeer, + HandshakedPeer +} import org.ergoplatform.network.PeerSpec import org.ergoplatform.settings.ErgoSettings import scorex.core.app.ScorexContext @@ -9,6 +14,7 @@ import scorex.core.network._ import scorex.core.utils.NetworkUtils import scorex.util.ScorexLogging +import scala.concurrent.duration._ import scala.util.Random /** @@ -20,13 +26,34 @@ class PeerManager(settings: ErgoSettings, scorexContext: ScorexContext) extends import PeerManager.ReceivableMessages._ private val peerDatabase = new PeerDatabase(settings) + private var connectedPeerAddresses = Set.empty[InetSocketAddress] + + override def preStart(): Unit = { + context.system.eventStream.subscribe(self, classOf[HandshakedPeer]) + context.system.eventStream.subscribe(self, classOf[DisconnectedPeer]) + scheduleOldPeersCleanup() + } + + override def postStop(): Unit = { + peerDatabase.close() + super.postStop() + } + + private def scheduleOldPeersCleanup(): Unit = { + context.system.scheduler.scheduleWithFixedDelay( + FiniteDuration(PeerDatabase.KnownPeerCleanupIntervalMs, MILLISECONDS), + FiniteDuration(PeerDatabase.KnownPeerCleanupIntervalMs, MILLISECONDS), + self, + CleanupOldPeers + )(context.system.dispatcher) + } if (peerDatabase.isEmpty) { // fill database with peers from config file if empty log.info("No peers in database, seeding peers database with nodes from config") settings.scorexSettings.network.knownPeers.foreach { address => if (!isSelf(address)) { - peerDatabase.addOrUpdateKnownPeer(PeerInfo.fromAddress(address)) + peerDatabase.addOrUpdateKnownPeer(PeerInfo.fromAddress(address), connectedPeerAddresses) } } } else { @@ -51,9 +78,24 @@ class PeerManager(settings: ErgoSettings, scorexContext: ScorexContext) extends case AddOrUpdatePeer(peerInfo) => // We have connected to a peer and got his peerInfo from him if (!isSelf(peerInfo.peerSpec) && !peerInfo.peerSpec.address.exists(isLocal(_))) { - peerDatabase.addOrUpdateKnownPeer(peerInfo) + peerDatabase.addOrUpdateKnownPeer(peerInfo, connectedPeerAddresses) } + case CleanupOldPeers => + peerDatabase.removeOldPeers(connectedPeerAddresses) + + case HandshakedPeer(remote) => + // Track both the transport endpoint and the advertised address (the database + // key). For inbound connections they differ: the socket address carries the + // peer's ephemeral source port, while the database is keyed by the advertised + // listening address from the handshake. Both must be protected from eviction. + connectedPeerAddresses += remote.connectionId.remoteAddress + remote.peerInfo.flatMap(_.peerSpec.address).foreach(connectedPeerAddresses += _) + + case DisconnectedPeer(connectedPeer) => + connectedPeerAddresses -= connectedPeer.connectionId.remoteAddress + connectedPeer.peerInfo.flatMap(_.peerSpec.address).foreach(connectedPeerAddresses -= _) + case Penalize(peer, penaltyType) => log.info(s"$peer penalized, penalty: $penaltyType") if (peerDatabase.penalize(peer, penaltyType)) { @@ -67,7 +109,7 @@ class PeerManager(settings: ErgoSettings, scorexContext: ScorexContext) extends if (peerSpec.address.forall(a => peerDatabase.get(a).isEmpty) && !isSelf(peerSpec) && !peerSpec.address.exists(isLocal(_))) { val peerInfo: PeerInfo = PeerInfo(peerSpec, 0, None) log.info(s"New discovered peer: $peerInfo") - peerDatabase.addOrUpdateKnownPeer(peerInfo) + peerDatabase.addOrUpdateKnownPeer(peerInfo, connectedPeerAddresses) } case RemovePeer(address) => @@ -125,6 +167,8 @@ object PeerManager { case class RemovePeer(address: InetSocketAddress) + case object CleanupOldPeers + /** * Message to get peers from known peers map filtered by `choose` function */ @@ -138,31 +182,52 @@ object PeerManager { * Choose at most `howMany` random peers, which were connected to our peer and weren't blacklisted. * * Used in peer propagation: peers chosen are recommended to a peer asking our node about more peers. + * + * Note: only a bounded window of the database is scanned. If that window happens + * to contain no eligible peers, the result is empty even when eligible peers exist + * elsewhere in the database. This is an accepted bounded-work tradeoff. */ case class SeenPeers(howMany: Int) extends GetPeers[Seq[PeerInfo]] with ScorexLogging { - val limit: Long = 3 * 60 * 60 * 1000 // 3h + val limit: Long = 3.hours.toMillis // 3h + + private val ScanBudgetMultiplier = 8 + private val MinScanBudget = 256 override def choose(knownPeers: Map[InetSocketAddress, PeerInfo], blacklistedPeers: Seq[InetAddress], sc: ScorexContext): Seq[PeerInfo] = { - val nonBlacklisted = knownPeers.values.toSeq - .filter { p => - (p.connectionType.isDefined || p.lastHandshake > 0) && - !blacklistedPeers.exists(ip => p.peerSpec.declaredAddress.exists(_.getAddress == ip)) - } - - val recentlySeenNonBlacklisted = nonBlacklisted.filter { p => - (System.currentTimeMillis() - p.lastStoredActivityTime < limit) - } - - if (recentlySeenNonBlacklisted.nonEmpty) { - val res = Random.shuffle(recentlySeenNonBlacklisted).take(howMany) - log.debug(s"Sending ${res.length} active peers: " + res) - res + if (howMany <= 0 || knownPeers.isEmpty) { + Seq.empty } else { - val res = Random.shuffle(nonBlacklisted).take(howMany) - log.debug(s"Sending ${res.length} known peers: " + res) + val scanBudget = math.max(howMany * ScanBudgetMultiplier, MinScanBudget) + val size = knownPeers.size + val window = math.min(scanBudget, size) + val start = + if (window == size) { + 0 + } else { + ThreadLocalRandom.current().nextInt(size - window + 1) + } + + val now = System.currentTimeMillis() + val cutoff = now - limit + + def isBlacklisted(p: PeerInfo): Boolean = + blacklistedPeers.exists(ip => p.peerSpec.declaredAddress.exists(_.getAddress == ip)) + + val candidates = knownPeers.valuesIterator + .drop(start) + .take(window) + .toSeq + .filter { p => + (p.connectionType.isDefined || p.lastHandshake > 0) && !isBlacklisted(p) + } + + val recentCandidates = candidates.filter(_.lastStoredActivityTime > cutoff) + val chosen = if (recentCandidates.nonEmpty) recentCandidates else candidates + val res = Random.shuffle(chosen).take(howMany) + log.debug(s"Sending ${res.length} peers (scanned $window of $size, window $start-${start + window})") res } } diff --git a/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/PopowProcessor.scala b/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/PopowProcessor.scala index 59922347a3..5e2ebd2183 100644 --- a/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/PopowProcessor.scala +++ b/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/PopowProcessor.scala @@ -107,8 +107,9 @@ trait PopowProcessor extends BasicReaders with ScorexLogging { * @return PoPow proof if success, Failure instance otherwise */ def popowProof(m: Int, k: Int, headerIdOpt: Option[ModifierId]): Try[NipopowProof] = { - val proofParams = PoPowParams(m, k, continuous = true) - NipopowProverWithDbAlgs.prove(historyReader, headerIdOpt = headerIdOpt, chainSettings)(proofParams) + PoPowParams(m, k, continuous = true).flatMap { proofParams => + NipopowProverWithDbAlgs.prove(historyReader, headerIdOpt = headerIdOpt, chainSettings)(proofParams) + } } /** diff --git a/src/main/scala/org/ergoplatform/nodeView/mempool/ErgoMemPool.scala b/src/main/scala/org/ergoplatform/nodeView/mempool/ErgoMemPool.scala index 6e58782dfe..a919afb76a 100644 --- a/src/main/scala/org/ergoplatform/nodeView/mempool/ErgoMemPool.scala +++ b/src/main/scala/org/ergoplatform/nodeView/mempool/ErgoMemPool.scala @@ -323,7 +323,8 @@ class ErgoMemPool private[mempool](private[mempool] val pool: OrderedTxPool, case _ => None } - loop(waitMinutes = 0).getOrElse(settings.nodeSettings.minimalFeeAmount) + val recommendedFee = loop(waitMinutes = 0).getOrElse(settings.nodeSettings.minimalFeeAmount) + math.max(recommendedFee, settings.nodeSettings.minimalFeeAmount) } /** @@ -346,8 +347,9 @@ class ErgoMemPool private[mempool](private[mempool] val pool: OrderedTxPool, // Time since statistics measurement interval (needed to calculate average tx rate) val elapsed = System.currentTimeMillis() - stats.startMeasurement - if (stats.takenTxns != 0) { - elapsed * posInPool / stats.takenTxns + val cappedElapsed = math.max(0L, math.min(elapsed, MemPoolStatistics.measurementIntervalMsec.toLong)) + if (stats.takenTxns > 0) { + cappedElapsed * posInPool / stats.takenTxns } else { 0 } diff --git a/src/main/scala/org/ergoplatform/nodeView/mempool/MemPoolStatistics.scala b/src/main/scala/org/ergoplatform/nodeView/mempool/MemPoolStatistics.scala index c88f8415d2..94d5db469b 100644 --- a/src/main/scala/org/ergoplatform/nodeView/mempool/MemPoolStatistics.scala +++ b/src/main/scala/org/ergoplatform/nodeView/mempool/MemPoolStatistics.scala @@ -51,7 +51,7 @@ case class MemPoolStatistics(startMeasurement: Long, object MemPoolStatistics { // Time parameters of mempool statistics val nHistogramBins: Int = 60 /* one hour */ - val measurementIntervalMsec: Int = 60 * 1000 /* one hour */ + val measurementIntervalMsec: Int = 60 * 1000 /* one minute */ val defaultPoolHistogram: List[FeeHistogramBin] = List.fill(MemPoolStatistics.nHistogramBins)(FeeHistogramBin(0, 0)) } diff --git a/src/main/scala/org/ergoplatform/nodeView/mempool/OrderedTxPool.scala b/src/main/scala/org/ergoplatform/nodeView/mempool/OrderedTxPool.scala index 2d5527ec7a..cc26879de1 100644 --- a/src/main/scala/org/ergoplatform/nodeView/mempool/OrderedTxPool.scala +++ b/src/main/scala/org/ergoplatform/nodeView/mempool/OrderedTxPool.scala @@ -50,6 +50,10 @@ class OrderedTxPool(val orderedTransactions: TreeMap[WeightedTxId, UnconfirmedTr case None if orderedTransactions.size == transactionsRegistry.size => orderedTransactions case _ => + log.warn( + s"Mempool indices diverged (ordered=${orderedTransactions.size}, " + + s"registry=${transactionsRegistry.size}); full scan to remove $id" + ) orderedTransactions.filter { case (wtx, utx) => wtx.id != id && utx.id != id } } } @@ -62,6 +66,7 @@ class OrderedTxPool(val orderedTransactions: TreeMap[WeightedTxId, UnconfirmedTr transactionsRegistry.get(id) .flatMap(wtx => orderedTransactions.get(wtx).filter(_.id == id).map(wtx -> _)) .orElse { + log.warn(s"Mempool fast lookup failed for $id, scanning ordered transactions") orderedTransactions.iterator.collectFirst { case (wtx, utx) if wtx.id == id && utx.id == id => wtx -> utx } diff --git a/src/main/scala/org/ergoplatform/nodeView/state/SnapshotsInfo.scala b/src/main/scala/org/ergoplatform/nodeView/state/SnapshotsInfo.scala index 990359091c..73ffbe28db 100644 --- a/src/main/scala/org/ergoplatform/nodeView/state/SnapshotsInfo.scala +++ b/src/main/scala/org/ergoplatform/nodeView/state/SnapshotsInfo.scala @@ -46,9 +46,9 @@ object SnapshotsInfoSerializer extends ErgoSerializer[SnapshotsInfo] { } override def parse(r: Reader): SnapshotsInfo = { - val manifestsCount = r.getUInt().toInt // we read from trusted source, no need for extra checks + val manifestsCount = r.getUIntExact() // we read from trusted source, no need for extra checks val manifests = (1 to manifestsCount).map { _ => - val h = r.getUInt().toInt + val h = r.getUIntExact() val manifestId = Digest32 @@ r.getBytes(Constants.HashLength) h -> manifestId }.toMap diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletSupport.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletSupport.scala index 03b91e19ca..17e2a7e36b 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletSupport.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletSupport.scala @@ -324,7 +324,7 @@ trait ErgoWalletSupport extends ScorexLogging { require(outputs.forall(_.additionalTokens.forall(_._2 > 0)), "Non-positive asset value") val assetIssueBox = outputs - .zip(requests) + .zip(requestsWithoutBurnTokens) .filter(_._2.isInstanceOf[AssetIssueRequest]) .map(_._1) .headOption diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/WalletTransaction.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/WalletTransaction.scala index 1ed83dab11..852154a5e2 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/WalletTransaction.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/WalletTransaction.scala @@ -6,7 +6,6 @@ import org.ergoplatform.wallet.Constants.ScanId import org.ergoplatform.serialization.ErgoSerializer import scorex.util.ModifierId import scorex.util.serialization.{Reader, Writer} -import scorex.util.Extensions._ import sigma.VersionContext /** @@ -57,7 +56,7 @@ object WalletTransactionSerializer extends ErgoSerializer[WalletTransaction] { (0 until scansCount).map(_ => ScanId @@ r.getShort()) } - val txBytesLen = r.getUInt().toIntExact + val txBytesLen = r.getUIntExact() // we use max supported script/tree to always parse everything there val tx = (VersionContext.withVersions(VersionContext.MaxSupportedScriptVersion, VersionContext.MaxSupportedScriptVersion) { ErgoTransactionSerializer.parseBytes(r.getBytes(txBytesLen)) diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletDigest.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletDigest.scala index 1b96025f1f..0c502e8c2b 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletDigest.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletDigest.scala @@ -4,7 +4,6 @@ import org.ergoplatform.nodeView.history.ErgoHistoryUtils._ import org.ergoplatform.nodeView.wallet.IdUtils._ import org.ergoplatform.settings.Constants import org.ergoplatform.serialization.ErgoSerializer -import scorex.util.Extensions._ import scorex.util.serialization.{Reader, Writer} import sigmastate.eval.Extensions.ArrayByteOps @@ -42,10 +41,10 @@ object WalletDigestSerializer extends ErgoSerializer[WalletDigest] { } override def parse(r: Reader): WalletDigest = { - val height = r.getUInt().toIntExact + val height = r.getUIntExact() val balance = r.getULong() - val walletAssetBalancesSize = r.getUInt().toIntExact + val walletAssetBalancesSize = r.getUIntExact() val walletAssetBalances = mutable.LinkedHashMap.empty[EncodedTokenId, Long] (0 until walletAssetBalancesSize).foreach { _ => diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/scanning/ScanningPredicateSerializer.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/scanning/ScanningPredicateSerializer.scala index ec1a6b4dda..ea1f0e0694 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/scanning/ScanningPredicateSerializer.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/scanning/ScanningPredicateSerializer.scala @@ -3,7 +3,6 @@ package org.ergoplatform.nodeView.wallet.scanning import org.ergoplatform.ErgoBox import org.ergoplatform.ErgoBox.RegisterId import org.ergoplatform.serialization.ErgoSerializer -import scorex.util.Extensions._ import scorex.util.serialization.{Reader, Writer} import sigma.ast.{EvaluatedValue, SType} import sigma.serialization.ValueSerializer @@ -55,7 +54,7 @@ object ScanningPredicateSerializer extends ErgoSerializer[ScanningPredicate] { def parseRegisterAndBytes(r: Reader): (RegisterId, EvaluatedValue[_ <: SType]) = { val reg = ErgoBox.registerByIndex(r.getByte()) - val len = r.getUInt().toIntExact + val len = r.getUIntExact() val bs = r.getBytes(len) val vs = ValueSerializer.deserialize(bs) reg -> vs.asInstanceOf[EvaluatedValue[SType]] diff --git a/src/main/scala/scorex/core/network/NetworkController.scala b/src/main/scala/scorex/core/network/NetworkController.scala index 8db3500db9..0cec487f35 100644 --- a/src/main/scala/scorex/core/network/NetworkController.scala +++ b/src/main/scala/scorex/core/network/NetworkController.scala @@ -163,11 +163,11 @@ class NetworkController(ergoSettings: ErgoSettings, peerManagerRef ! PeerManager.ReceivableMessages.Penalize(peerAddress, penaltyType) case Blacklisted(peerAddress) => - connections.get(peerAddress).foreach { peer => - connections = connections.filterNot { case (address, _) => // clear all connections related to banned peer ip - Option(peer.connectionId.remoteAddress.getAddress).exists(Option(address.getAddress).contains(_)) - } - peer.handlerRef ! CloseConnection + Option(peerAddress.getAddress).foreach { blacklistedIp => + val peersToClose = connections.valuesIterator.filter { peer => + Option(peer.connectionId.remoteAddress.getAddress).contains(blacklistedIp) + }.toSeq + peersToClose.foreach(_.handlerRef ! CloseConnection) } } diff --git a/src/main/scala/scorex/core/network/PeerConnectionHandler.scala b/src/main/scala/scorex/core/network/PeerConnectionHandler.scala index 8c32487397..bd6a573bed 100644 --- a/src/main/scala/scorex/core/network/PeerConnectionHandler.scala +++ b/src/main/scala/scorex/core/network/PeerConnectionHandler.scala @@ -4,14 +4,19 @@ import akka.actor.{Actor, ActorRef, Cancellable, Props, SupervisorStrategy} import akka.io.Tcp import akka.io.Tcp._ import akka.util.{ByteString, CompactByteString} -import org.ergoplatform.network.{Handshake, HandshakeSerializer, PeerSpec, Version} import org.ergoplatform.network.Version.Eip37ForkVersion -import scorex.core.app.ScorexContext -import scorex.core.network.NetworkController.ReceivableMessages.{Handshaked, PenalizePeer} -import scorex.core.network.PeerConnectionHandler.ReceivableMessages +import org.ergoplatform.network.{Handshake, HandshakeSerializer, PeerSpec, Version} +import org.ergoplatform.network.message.MessageConstants.{ + ChecksumLength, + HeaderLength, + MaxMessageSize +} import org.ergoplatform.network.message.MessageSerializer import org.ergoplatform.network.peer.{PeerInfo, PenaltyType} import org.ergoplatform.settings.ScorexSettings +import scorex.core.app.ScorexContext +import scorex.core.network.NetworkController.ReceivableMessages.{Handshaked, PenalizePeer} +import scorex.core.network.PeerConnectionHandler.ReceivableMessages import scorex.util.ScorexLogging import scala.annotation.tailrec @@ -27,6 +32,7 @@ class PeerConnectionHandler(scorexSettings: ScorexSettings, extends Actor with ScorexLogging { import PeerConnectionHandler.ReceivableMessages._ + import PeerConnectionHandler.{MaxBufferedOutboundBytes, MaxBufferedOutboundMessages} private val networkSettings = scorexSettings.network private val connection = connectionDescription.connection @@ -48,6 +54,8 @@ class PeerConnectionHandler(scorexSettings: ScorexSettings, private var outMessagesBuffer: TreeMap[Long, ByteString] = TreeMap.empty + private var outMessagesBufferBytes: Long = 0L + private var outMessagesCounter: Long = 0 override def preStart: Unit = { @@ -179,7 +187,10 @@ class PeerConnectionHandler(scorexSettings: ScorexSettings, writeFirst() case ReceivableMessages.Ack(id) => - outMessagesBuffer -= id + outMessagesBuffer.get(id).foreach { msg => + outMessagesBuffer -= id + outMessagesBufferBytes -= msg.length + } if (outMessagesBuffer.nonEmpty){ writeFirst() } else { @@ -226,7 +237,22 @@ class PeerConnectionHandler(scorexSettings: ScorexSettings, } private def buffer(id: Long, msg: ByteString): Unit = { - outMessagesBuffer += id -> msg + val previousMessage = outMessagesBuffer.get(id) + val previousLength = previousMessage.fold(0)(_.length) + val candidateBytes = outMessagesBufferBytes - previousLength + msg.length + val candidateMessages = outMessagesBuffer.size + previousMessage.fold(1)(_ => 0) + if (candidateBytes > MaxBufferedOutboundBytes || + candidateMessages > MaxBufferedOutboundMessages) { + log.warn(s"Buffered outbound data for $connectionId would exceed its limit " + + s"($candidateMessages messages, $candidateBytes bytes), aborting the connection") + outMessagesBuffer = TreeMap.empty + outMessagesBufferBytes = 0L + connection ! Abort + context.stop(self) + } else { + outMessagesBuffer += id -> msg + outMessagesBufferBytes = candidateBytes + } } private def writeFirst(): Unit = { @@ -259,6 +285,14 @@ class PeerConnectionHandler(scorexSettings: ScorexSettings, object PeerConnectionHandler { + // Keep one maximum serialized frame per peer. Backpressured snapshot transfers + // retry instead of retaining their entire application-level in-flight window. + private[network] val MaxBufferedOutboundBytes: Long = + MaxMessageSize.toLong + HeaderLength + ChecksumLength + + // Independently bound collection overhead from small messages. + private[network] val MaxBufferedOutboundMessages: Int = 64 + object ReceivableMessages { case object HandshakeTimeout diff --git a/src/test/scala/org/ergoplatform/http/routes/ErgoHttpServiceSpec.scala b/src/test/scala/org/ergoplatform/http/routes/ErgoHttpServiceSpec.scala new file mode 100644 index 0000000000..13b730ecc8 --- /dev/null +++ b/src/test/scala/org/ergoplatform/http/routes/ErgoHttpServiceSpec.scala @@ -0,0 +1,112 @@ +package org.ergoplatform.http.routes + +import akka.http.scaladsl.model.StatusCodes +import akka.http.scaladsl.server.Route +import akka.http.scaladsl.testkit.ScalatestRouteTest +import ch.qos.logback.classic.spi.ILoggingEvent +import ch.qos.logback.classic.{Level, Logger => LogbackLogger} +import ch.qos.logback.core.read.ListAppender +import org.ergoplatform.http.api.EmissionApiRoute +import org.ergoplatform.http.{ErgoHttpService, NodePanelRoute, SwaggerRoute} +import org.ergoplatform.utils.Stubs +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.slf4j.LoggerFactory + +import scala.collection.JavaConverters._ + +class ErgoHttpServiceSpec extends AnyFlatSpec + with Matchers + with ScalatestRouteTest + with Stubs { + + import org.ergoplatform.utils.ErgoNodeTestConstants._ + + private val restApiSettings = settings.scorexSettings.restApi + + private val service = ErgoHttpService( + apiRoutes = Seq(EmissionApiRoute(settings)), + swaggerRoute = SwaggerRoute(restApiSettings, swaggerConfig = ""), + panelRoute = NodePanelRoute() + ) + + private val route: Route = service.compositeRoute + + private val serviceLogger: LogbackLogger = + LoggerFactory.getLogger(classOf[ErgoHttpService]).asInstanceOf[LogbackLogger] + + /** Runs `body` while capturing what the service logs at `level` */ + private def capturingLogs[T](level: Level)(body: => T): (T, Seq[String]) = { + val appender = new ListAppender[ILoggingEvent] + appender.start() + val previousLevel = serviceLogger.getLevel + serviceLogger.setLevel(level) + serviceLogger.addAppender(appender) + try { + val result = body + (result, appender.list.asScala.map(_.getFormattedMessage).toList) + } finally { + serviceLogger.detachAppender(appender) + serviceLogger.setLevel(previousLevel) + appender.stop() + } + } + + it should "log served queries at DEBUG level" in { + val (_, messages) = capturingLogs(Level.DEBUG) { + Get("/emission/at/100") ~> route ~> check { + status shouldBe StatusCodes.OK + } + } + + val logged = messages.filter(_.startsWith("GET /emission/at/100")) + logged.size shouldBe 1 + // method, uri, response status and elapsed time, and nothing else + logged.head should fullyMatch regex """GET /emission/at/100 - 200 in \d+ ms""" + } + + it should "log the query string, and log unmatched paths with the status they were rejected with" in { + val (rejectedStatus, messages) = capturingLogs(Level.DEBUG) { + Get("/emission/at/100?foo=bar") ~> route ~> check { + status shouldBe StatusCodes.OK + } + Get("/no/such/route") ~> route ~> check { + status.isSuccess() shouldBe false + status.intValue() + } + } + + messages.exists(_.startsWith("GET /emission/at/100?foo=bar - 200 in ")) shouldBe true + // rejections are turned into responses by the rejection handler, so they are logged too + messages.exists(_.startsWith(s"GET /no/such/route - $rejectedStatus in ")) shouldBe true + } + + it should "log nothing when the logger is not at DEBUG level" in { + val (_, messages) = capturingLogs(Level.INFO) { + Get("/emission/at/100") ~> route ~> check { + status shouldBe StatusCodes.OK + } + } + + messages shouldBe empty + } + + it should "not change the response when logging is enabled" in { + val body = capturingLogs(Level.DEBUG) { + Get("/emission/at/100") ~> route ~> check { + status shouldBe StatusCodes.OK + responseAs[String] + } + }._1 + + val bodyWithoutLogging = capturingLogs(Level.OFF) { + Get("/emission/at/100") ~> route ~> check { + status shouldBe StatusCodes.OK + responseAs[String] + } + }._1 + + body shouldBe bodyWithoutLogging + } + +} diff --git a/src/test/scala/org/ergoplatform/http/routes/ErgoPeersApiRouteSpec.scala b/src/test/scala/org/ergoplatform/http/routes/ErgoPeersApiRouteSpec.scala index 5353ec7499..fc739ae2df 100644 --- a/src/test/scala/org/ergoplatform/http/routes/ErgoPeersApiRouteSpec.scala +++ b/src/test/scala/org/ergoplatform/http/routes/ErgoPeersApiRouteSpec.scala @@ -15,6 +15,7 @@ import org.scalatest.matchers.should.Matchers import org.scalatestplus.scalacheck.ScalaCheckPropertyChecks import scorex.core.network.NetworkController.ReceivableMessages.GetConnectedPeers import org.ergoplatform.network.peer.PeerManager.ReceivableMessages.GetAllPeers +import org.ergoplatform.network.peer.PeerInfo import org.ergoplatform.settings.RESTApiSettings import java.net.InetSocketAddress @@ -89,5 +90,23 @@ class ErgoPeersApiRouteSpec extends AnyFlatSpec } } } + + it should "return all known peers" in { + val networkControllerProbe = TestProbe() + val route: Route = ErgoPeersApiRoute(peerManagerProbe.ref, networkControllerProbe.ref, null, null, restApiSettings).route + val peers = (1 to 55).map { i => + val addr = new InetSocketAddress(s"8.8.0.$i", 9000 + i) + addr -> PeerInfo.fromAddress(addr) + }.toMap + Future { + peerManagerProbe.expectMsg(GetAllPeers) + peerManagerProbe.reply(peers) + } + + Get("/peers/all") ~> route ~> check { + status shouldBe StatusCodes.OK + responseAs[Json].asArray.get.size shouldBe 55 + } + } } diff --git a/src/test/scala/org/ergoplatform/http/routes/NipopowApiRoutesSpec.scala b/src/test/scala/org/ergoplatform/http/routes/NipopowApiRoutesSpec.scala index cfc95a8337..205aedc0bf 100644 --- a/src/test/scala/org/ergoplatform/http/routes/NipopowApiRoutesSpec.scala +++ b/src/test/scala/org/ergoplatform/http/routes/NipopowApiRoutesSpec.scala @@ -35,6 +35,12 @@ class NipopowApiRoutesSpec extends AnyFlatSpec } } + it should "reject proof request when minimum and suffix length overflow" in { + Get(s"/nipopow/proof/${Int.MaxValue}/1") ~> route ~> check { + status shouldBe StatusCodes.BadRequest + } + } + it should "proof request with missing headerId" in { Get("/nipopow/proof/1/1/05bf63aa1ecfc9f4e3fadc993f87b33edb4d58e151c1891816d734dd5a0e2e09") ~> route ~> check { status shouldBe StatusCodes.BadRequest diff --git a/src/test/scala/org/ergoplatform/local/NipopowVerifierSpec.scala b/src/test/scala/org/ergoplatform/local/NipopowVerifierSpec.scala index 1216d77244..e4b8543b8d 100644 --- a/src/test/scala/org/ergoplatform/local/NipopowVerifierSpec.scala +++ b/src/test/scala/org/ergoplatform/local/NipopowVerifierSpec.scala @@ -1,5 +1,8 @@ package org.ergoplatform.local +import java.util.concurrent.{CountDownLatch, TimeUnit} +import java.util.concurrent.atomic.AtomicReference + import org.ergoplatform.modifiers.history.popow.{PoPowHeader, PoPowParams} import org.ergoplatform.modifiers.ErgoFullBlock import org.scalatest.matchers.should.Matchers @@ -11,7 +14,7 @@ class NipopowVerifierSpec extends AnyPropSpec with Matchers { import org.ergoplatform.utils.generators.ChainGenerator._ - private val poPowParams = PoPowParams(30, 30, continuous = false) + private val poPowParams = PoPowParams(30, 30, continuous = false).get val toPoPoWChain = (c: Seq[ErgoFullBlock]) => c.map(b => PoPowHeader.fromBlock(b).get) property("processes new proofs") { @@ -43,4 +46,49 @@ class NipopowVerifierSpec extends AnyPropSpec with Matchers { verifier.bestChain.last.id shouldBe longestProof.headersChain.last.id } } + + property("rejects proofs with invalid security parameters") { + val baseChain = genChain(100) + val params = PoPowParams(5, 5, continuous = false).get + val proof = nipopowAlgos.prove(toPoPoWChain(baseChain))(params).get + + Seq( + proof.copy(m = 0), + proof.copy(k = 0), + proof.copy(m = Int.MaxValue, k = 1) + ).foreach { invalidProof => + val proofBytes = invalidProof.serializer.toBytes(invalidProof) + val receivedProof = invalidProof.serializer.parseBytes(proofBytes) + receivedProof.isValid shouldBe false + + val verifier = new NipopowVerifier(Some(baseChain.head.id)) + verifier.process(receivedProof) shouldBe ValidationError + verifier.bestChain shouldBe empty + } + } + + property("returns when a duplicate invalid proof is processed") { + val baseChain = genChain(100) + val params = PoPowParams(5, 5, continuous = false).get + val invalidProof = nipopowAlgos.prove(toPoPoWChain(baseChain))(params).get.copy(m = 0) + val proofBytes = invalidProof.serializer.toBytes(invalidProof) + val receivedProof = invalidProof.serializer.parseBytes(proofBytes) + val verifier = new NipopowVerifier(Some(baseChain.head.id)) + + val firstResult = verifier.process(receivedProof) + val secondResult = new AtomicReference[NipopowProofVerificationResult]() + val completed = new CountDownLatch(1) + val worker = new Thread(new Runnable { + override def run(): Unit = + try secondResult.set(verifier.process(receivedProof)) + finally completed.countDown() + }) + worker.setDaemon(true) + worker.start() + + completed.await(2, TimeUnit.SECONDS) shouldBe true + firstResult shouldBe ValidationError + secondResult.get() shouldBe ValidationError + verifier.bestChain shouldBe empty + } } diff --git a/src/test/scala/org/ergoplatform/mining/CandidateGeneratorSpec.scala b/src/test/scala/org/ergoplatform/mining/CandidateGeneratorSpec.scala index 26787c525c..c049a80872 100644 --- a/src/test/scala/org/ergoplatform/mining/CandidateGeneratorSpec.scala +++ b/src/test/scala/org/ergoplatform/mining/CandidateGeneratorSpec.scala @@ -151,6 +151,70 @@ class CandidateGeneratorSpec extends AnyFlatSpec with Matchers with ErgoTestHelp system.terminate() } + it should "recover when locally mined block is invalidated by full block id" in new TestKit( + ActorSystem() + ) { + val replyProbe = new TestProbe(system) + // fake node view holder: solved block is never applied, so solvedBlock stays set + val viewHolderProbe = new TestProbe(system) + + // real readers holder over real node view holder, needed for candidate generation + val realViewHolderRef: ActorRef = ErgoNodeViewRef(defaultSettings) + val readersHolderRef: ActorRef = ErgoReadersHolderRef(realViewHolderRef) + + val candidateGenerator: ActorRef = + CandidateGenerator( + defaultMinerSecret.publicImage, + readersHolderRef, + viewHolderProbe.ref, + defaultSettings + ) + + candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = false), replyProbe.ref) + val block = replyProbe.expectMsgPF(candidateGenDelay) { + case StatusReply.Success(candidate: Candidate) => + defaultSettings.chainSettings.powScheme + .proveCandidate(candidate.candidateBlock, defaultMinerSecret.w, 0, 1000) + .get + } + + candidateGenerator.tell(block.header.powSolution, replyProbe.ref) + replyProbe.expectMsg(blockValidationDelay, StatusReply.Success(())) + + // block sections were sent to the (fake) node view holder + viewHolderProbe.expectMsg(LocallyGeneratedModifier(block.header)) + block.mandatoryBlockSections.foreach { section => + viewHolderProbe.expectMsg(LocallyGeneratedModifier(section)) + } + + // mining is stalled: new solutions are rejected while solvedBlock is set + candidateGenerator.tell(block.header.powSolution, replyProbe.ref) + replyProbe.expectMsgPF(blockValidationDelay) { + case r: StatusReply[_] if r.isError => + } + + // node view holder invalidates the block using full-block typeId and block id + val failedTxId = block.blockTransactions.txs.head.id + val error = + new MalformedModifierError("tx failed", failedTxId, ErgoTransaction.modifierTypeId) + system.eventStream.publish( + SemanticallyFailedModification(ErgoFullBlock.modifierTypeId, block.id, error) + ) + + // mining resumes: a new candidate is generated and new solutions are accepted again + candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = false), replyProbe.ref) + val newBlock = replyProbe.expectMsgPF(candidateGenDelay) { + case StatusReply.Success(candidate: Candidate) => + defaultSettings.chainSettings.powScheme + .proveCandidate(candidate.candidateBlock, defaultMinerSecret.w, 0, 1000) + .get + } + candidateGenerator.tell(newBlock.header.powSolution, replyProbe.ref) + replyProbe.expectMsg(blockValidationDelay, StatusReply.Success(())) + + system.terminate() + } + it should "let multiple miners compete" in new TestKit(ActorSystem()) { val testProbe = new TestProbe(system) system.eventStream.subscribe(testProbe.ref, newBlockSignal) diff --git a/src/test/scala/org/ergoplatform/modifiers/history/PoPowAlgosSpec.scala b/src/test/scala/org/ergoplatform/modifiers/history/PoPowAlgosSpec.scala index 66d903e79b..54c01f2ecb 100644 --- a/src/test/scala/org/ergoplatform/modifiers/history/PoPowAlgosSpec.scala +++ b/src/test/scala/org/ergoplatform/modifiers/history/PoPowAlgosSpec.scala @@ -1,5 +1,8 @@ package org.ergoplatform.modifiers.history +import java.util.concurrent.{CountDownLatch, TimeUnit} +import java.util.concurrent.atomic.AtomicReference + import org.ergoplatform.modifiers.history.popow.{NipopowAlgos, NipopowProof, PoPowHeader, PoPowParams} import org.ergoplatform.modifiers.ErgoFullBlock import org.scalacheck.Gen @@ -12,11 +15,45 @@ class PoPowAlgosSpec extends AnyPropSpec with Matchers { import org.ergoplatform.utils.generators.CoreObjectGenerators._ import org.ergoplatform.utils.ErgoCoreTestConstants._ - private val poPowParams = PoPowParams(30, 30, continuous = false) + private val poPowParams = PoPowParams(30, 30, continuous = false).get private val ChainLength = 10 private def toPoPoWChain = (c: Seq[ErgoFullBlock]) => c.map(b => PoPowHeader.fromBlock(b).get) + property("PoPowParams rejects invalid minimum chain lengths") { + PoPowParams.isValid(0, 1) shouldBe false + PoPowParams.isValid(1, 0) shouldBe false + PoPowParams.isValid(Int.MaxValue, 1) shouldBe false + + PoPowParams(0, 1, continuous = false) shouldBe 'failure + PoPowParams(1, 0, continuous = false) shouldBe 'failure + PoPowParams(Int.MaxValue, 1, continuous = false) shouldBe 'failure + + PoPowParams.isValid(Int.MaxValue - 1, 1) shouldBe true + PoPowParams(1, 1, continuous = false).get.minChainLength shouldBe 2 + } + + property("bestArg rejects a non-positive security parameter without looping") { + val algos = nipopowAlgos + val completed = new CountDownLatch(1) + val error = new AtomicReference[Throwable]() + val worker = new Thread(new Runnable { + override def run(): Unit = + try { + algos.bestArg(Seq.empty)(0) + } catch { + case t: Throwable => error.set(t) + } finally { + completed.countDown() + } + }) + worker.setDaemon(true) + worker.start() + + completed.await(2, TimeUnit.SECONDS) shouldBe true + error.get() shouldBe a[IllegalArgumentException] + } + property("updateInterlinks") { val chain = genChain(ChainLength) val genesis = chain.head @@ -144,7 +181,7 @@ class PoPowAlgosSpec extends AnyPropSpec with Matchers { } property("isBetterThan - a disconnected prefix chain should not win") { - val smallPoPowParams = PoPowParams(50, 1, continuous = false) + val smallPoPowParams = PoPowParams(50, 1, continuous = false).get val size = 100 val chain = toPoPoWChain(genChain(size)) val proof = nipopowAlgos.prove(chain)(smallPoPowParams).get @@ -158,7 +195,7 @@ class PoPowAlgosSpec extends AnyPropSpec with Matchers { } property("hasValidConnections - ensures a connected prefix chain") { - val smallPoPowParams = PoPowParams(5, 5, continuous = false) + val smallPoPowParams = PoPowParams(5, 5, continuous = false).get val sizes = Seq(100, 200) sizes.foreach { size => val chain = toPoPoWChain(genChain(size)) @@ -172,7 +209,7 @@ class PoPowAlgosSpec extends AnyPropSpec with Matchers { } property("hasValidConnections - ensures a connected suffix chain") { - val smallPoPowParams = PoPowParams(5, 5, continuous = false) + val smallPoPowParams = PoPowParams(5, 5, continuous = false).get val sizes = Seq(100, 200) sizes.foreach { size => diff --git a/src/test/scala/org/ergoplatform/modifiers/history/PoPowAlgosWithDBSpec.scala b/src/test/scala/org/ergoplatform/modifiers/history/PoPowAlgosWithDBSpec.scala index c81df767c2..f708e6e966 100644 --- a/src/test/scala/org/ergoplatform/modifiers/history/PoPowAlgosWithDBSpec.scala +++ b/src/test/scala/org/ergoplatform/modifiers/history/PoPowAlgosWithDBSpec.scala @@ -12,7 +12,7 @@ class PoPowAlgosWithDBSpec extends AnyPropSpec with Matchers { import org.ergoplatform.utils.generators.ChainGenerator._ property("proof(chain) is equivalent to proof(histReader)") { - val poPowParams = PoPowParams(m = 5, k = 6, continuous = false) + val poPowParams = PoPowParams(m = 5, k = 6, continuous = false).get val blocksChain = genChain(3000) val pchain = blocksChain.map(b => PoPowHeader.fromBlock(b).get) val proof0 = nipopowAlgos.prove(pchain)(poPowParams).get @@ -30,7 +30,7 @@ class PoPowAlgosWithDBSpec extends AnyPropSpec with Matchers { } property("proof(histReader) for a header in the past") { - val poPowParams = PoPowParams(5, 6, continuous = false) + val poPowParams = PoPowParams(5, 6, continuous = false).get val blocksChain = genChain(300) val at = 200 diff --git a/src/test/scala/org/ergoplatform/network/peer/PeerDatabaseSpec.scala b/src/test/scala/org/ergoplatform/network/peer/PeerDatabaseSpec.scala new file mode 100644 index 0000000000..98fc3486da --- /dev/null +++ b/src/test/scala/org/ergoplatform/network/peer/PeerDatabaseSpec.scala @@ -0,0 +1,259 @@ +package org.ergoplatform.network.peer + +import org.ergoplatform.db.DBSpec +import org.ergoplatform.network.PeerSpec +import org.ergoplatform.settings.ErgoSettings +import org.ergoplatform.utils.ErgoCorePropertyTest +import org.ergoplatform.utils.ErgoNodeTestConstants._ +import scorex.db.LDBFactory + +import java.io.File +import java.net.InetSocketAddress + +class PeerDatabaseSpec extends ErgoCorePropertyTest with DBSpec { + + private def testSettings(dir: File): ErgoSettings = + settings.copy(directory = dir.getAbsolutePath) + + private def peerInfo(address: InetSocketAddress, lastHandshake: Long): PeerInfo = { + PeerInfo( + defaultPeerSpec.copy(declaredAddress = Some(address)), + lastHandshake, + None, + 0L + ) + } + + private def peerInfo(spec: PeerSpec, lastHandshake: Long): PeerInfo = { + PeerInfo(spec, lastHandshake, None, 0L) + } + + private def withDb[T](maxKnownPeers: Int = PeerDatabase.MaxKnownPeers) + (body: PeerDatabase => T): T = { + val dir = createTempDir + val db = new PeerDatabase(testSettings(dir), maxKnownPeers) + try { + body(db) + } finally { + db.close() + deleteRecursive(dir) + } + } + + property("PeerDatabase should store and retrieve a known peer") { + val address = new InetSocketAddress("8.8.8.8", 9001) + val info = peerInfo(address, System.currentTimeMillis()) + withDb() { db => + db.addOrUpdateKnownPeer(info) + db.get(address) shouldBe Some(info) + db.knownPeers should contain(address -> info) + } + } + + property("PeerDatabase should ignore a peer without a usable address") { + val info = peerInfo(defaultPeerSpec, System.currentTimeMillis()) + withDb() { db => + db.addOrUpdateKnownPeer(info) + db.knownPeers shouldBe empty + } + } + + property("PeerDatabase should cap and evict oldest non-connected peer") { + val addresses = (1 to 4).map(i => new InetSocketAddress(s"8.8.8.$i", 9000 + i)) + withDb(maxKnownPeers = 3) { db => + addresses.zip(Seq(1L, 2L, 3L, 4L)).foreach { case (addr, ts) => + db.addOrUpdateKnownPeer(peerInfo(addr, ts)) + } + db.knownPeers.keys should contain(addresses(1)) + db.knownPeers.keys should contain(addresses(2)) + db.knownPeers.keys should contain(addresses(3)) + db.knownPeers.keys should not contain addresses(0) + } + } + + property("PeerDatabase should not evict a connected peer when making room") { + val addresses = (1 to 4).map(i => new InetSocketAddress(s"8.8.8.$i", 9000 + i)) + val connected = Set(addresses.head) + withDb(maxKnownPeers = 3) { db => + addresses.zip(Seq(1L, 2L, 3L, 4L)).foreach { case (addr, ts) => + db.addOrUpdateKnownPeer(peerInfo(addr, ts), connected) + } + db.knownPeers.keys should contain(addresses(0)) + db.knownPeers.keys should contain(addresses(2)) + db.knownPeers.keys should contain(addresses(3)) + db.knownPeers.keys should not contain addresses(1) + } + } + + property("PeerDatabase should ignore peer older than oldest when full") { + val addresses = (1 to 3).map(i => new InetSocketAddress(s"8.8.8.$i", 9000 + i)) + val older = new InetSocketAddress("8.8.8.100", 9999) + withDb(maxKnownPeers = 3) { db => + addresses.zip(Seq(10L, 20L, 30L)).foreach { case (addr, ts) => + db.addOrUpdateKnownPeer(peerInfo(addr, ts)) + } + db.addOrUpdateKnownPeer(peerInfo(older, 5L)) + db.knownPeers.keys should not contain older + } + } + + property("PeerDatabase should remove only old disconnected peers during cleanup") { + var connected = Set.empty[InetSocketAddress] + val oldConnected = new InetSocketAddress("8.8.8.1", 9001) + val oldDisconnected = new InetSocketAddress("8.8.8.2", 9002) + val recent = new InetSocketAddress("8.8.8.3", 9003) + val now = System.currentTimeMillis() + withDb(maxKnownPeers = 100) { db => + connected += oldConnected + val oldTs = now - PeerDatabase.KnownPeerMaxAgeMs - 1000 + db.addOrUpdateKnownPeer(peerInfo(oldConnected, oldTs), connected) + db.addOrUpdateKnownPeer(peerInfo(oldDisconnected, oldTs), connected) + db.addOrUpdateKnownPeer(peerInfo(recent, now - 1000), connected) + db.removeOldPeers(connected) + db.knownPeers.keys should contain(oldConnected) + db.knownPeers.keys should contain(recent) + db.knownPeers.keys should not contain oldDisconnected + } + } + + property("PeerDatabase should persist peers across close and reopen") { + val dir = createTempDir + val dbSettings = testSettings(dir) + val address = new InetSocketAddress("8.8.8.8", 9001) + val info = peerInfo(address, 123456789L) + try { + val db1 = new PeerDatabase(dbSettings) + db1.addOrUpdateKnownPeer(info) + db1.close() + val db2 = new PeerDatabase(dbSettings) + db2.get(address) shouldBe Some(info) + db2.knownPeers should contain(address -> info) + db2.close() + } finally { + deleteRecursive(dir) + } + } + + property("PeerDatabase should not reload removed peers") { + val dir = createTempDir + val dbSettings = testSettings(dir) + val address1 = new InetSocketAddress("8.8.8.1", 9001) + val address2 = new InetSocketAddress("8.8.8.2", 9002) + try { + val db1 = new PeerDatabase(dbSettings) + db1.addOrUpdateKnownPeer(peerInfo(address1, 100L)) + db1.addOrUpdateKnownPeer(peerInfo(address2, 200L)) + db1.remove(address1) + db1.close() + val db2 = new PeerDatabase(dbSettings) + db2.knownPeers.keys should not contain address1 + db2.knownPeers should contain(address2 -> peerInfo(address2, 200L)) + db2.close() + } finally { + deleteRecursive(dir) + } + } + + property("PeerDatabase should load only newest peers when persisted set exceeds cap") { + val dir = createTempDir + val dbSettings = testSettings(dir) + val addresses = (1 to 5).map(i => new InetSocketAddress(s"8.8.8.$i", 9000 + i)) + // timestamps permuted independently of insertion/address ordering: the newest + // timestamps belong to addresses(2), addresses(4) and addresses(0) + val timestamps = Map( + addresses(0) -> 40L, + addresses(1) -> 10L, + addresses(2) -> 50L, + addresses(3) -> 20L, + addresses(4) -> 30L + ) + try { + val db1 = new PeerDatabase(dbSettings, maxKnownPeers = 5) + addresses.foreach { addr => + db1.addOrUpdateKnownPeer(peerInfo(addr, timestamps(addr))) + } + db1.knownPeers should have size 5 + db1.close() + + // cap shrunk to 3: retention must be driven by timestamps, not store order + val db2 = new PeerDatabase(dbSettings, maxKnownPeers = 3) + db2.knownPeers should have size 3 + db2.knownPeers.keys should contain(addresses(0)) + db2.knownPeers.keys should contain(addresses(2)) + db2.knownPeers.keys should contain(addresses(4)) + db2.knownPeers.keys should not contain addresses(1) + db2.knownPeers.keys should not contain addresses(3) + db2.close() + + // dropped records must be physically deleted from the store: reopening with + // the original cap must not resurrect them + val db3 = new PeerDatabase(dbSettings, maxKnownPeers = 5) + db3.knownPeers should have size 3 + db3.knownPeers.keys should contain(addresses(0)) + db3.knownPeers.keys should contain(addresses(2)) + db3.knownPeers.keys should contain(addresses(4)) + db3.close() + } finally { + deleteRecursive(dir) + } + } + + property("PeerDatabase should physically remove malformed and oversized records on startup") { + val dir = createTempDir + val dbSettings = testSettings(dir) + val address = new InetSocketAddress("8.8.8.8", 9001) + val garbageKey = Array[Byte](1, 2, 3) + val garbageValue = Array[Byte](4, 5, 6) + val oversizedKey = new Array[Byte](PeerDatabase.MaxSerializedPeerAddressSize + 1) + val oversizedValue = new Array[Byte](PeerDatabase.MaxSerializedPeerInfoSize + 1) + try { + val db1 = new PeerDatabase(dbSettings) + db1.addOrUpdateKnownPeer(peerInfo(address, 100L)) + db1.close() + + // plant malformed records directly into the store + val rawStore = LDBFactory.createKvDb(s"${dir.getAbsolutePath}/peers") + rawStore.insert(garbageKey, garbageValue) + rawStore.insert(oversizedKey, oversizedValue) + rawStore.close() + + val db2 = new PeerDatabase(dbSettings) + // valid peer survives, malformed records are skipped + db2.knownPeers.keys should contain(address) + db2.knownPeers should have size 1 + db2.close() + + // malformed records are physically removed, not reparsed on every startup + val db3 = new PeerDatabase(dbSettings) + db3.knownPeers should have size 1 + db3.close() + + val checkStore = LDBFactory.createKvDb(s"${dir.getAbsolutePath}/peers") + checkStore.get(garbageKey) shouldBe empty + checkStore.get(oversizedKey) shouldBe empty + checkStore.close() + } finally { + deleteRecursive(dir) + } + } + + property("PeerDatabase should keep untried peers (zero lastHandshake) during cleanup") { + val untried = new InetSocketAddress("8.8.8.1", 9001) + val unavailableSeed = new InetSocketAddress("8.8.8.2", 9002) + val oldHandshaked = new InetSocketAddress("8.8.8.3", 9003) + val recent = new InetSocketAddress("8.8.8.4", 9004) + val now = System.currentTimeMillis() + withDb(maxKnownPeers = 100) { db => + db.addOrUpdateKnownPeer(peerInfo(untried, 0L)) + db.addOrUpdateKnownPeer(peerInfo(unavailableSeed, 0L)) + db.addOrUpdateKnownPeer(peerInfo(oldHandshaked, now - PeerDatabase.KnownPeerMaxAgeMs - 1000)) + db.addOrUpdateKnownPeer(peerInfo(recent, now - 1000)) + db.removeOldPeers() + db.knownPeers.keys should contain(untried) + db.knownPeers.keys should contain(unavailableSeed) + db.knownPeers.keys should contain(recent) + db.knownPeers.keys should not contain oldHandshaked + } + } + +} diff --git a/src/test/scala/org/ergoplatform/network/peer/PeerManagerSpec.scala b/src/test/scala/org/ergoplatform/network/peer/PeerManagerSpec.scala new file mode 100644 index 0000000000..c643130aeb --- /dev/null +++ b/src/test/scala/org/ergoplatform/network/peer/PeerManagerSpec.scala @@ -0,0 +1,239 @@ +package org.ergoplatform.network.peer + +import akka.actor.ActorRef +import akka.testkit.{TestActorRef, TestProbe} +import org.ergoplatform.db.DBSpec +import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages.{ + DisconnectedPeer, + HandshakedPeer +} +import org.ergoplatform.network.PeerSpec +import org.ergoplatform.settings.ErgoSettings +import org.ergoplatform.utils.ErgoCorePropertyTest +import org.ergoplatform.utils.ErgoNodeTestConstants._ +import scorex.core.app.ScorexContext +import scorex.core.network.{ConnectionDirection, ConnectionId, ConnectedPeer, Incoming, Outgoing} +import scorex.testkit.utils.AkkaFixture + +import java.io.File +import java.net.{InetAddress, InetSocketAddress} +import scala.concurrent.Await +import scala.concurrent.duration.Duration + +class PeerManagerSpec extends ErgoCorePropertyTest with DBSpec { + + import PeerManager.ReceivableMessages._ + + private class PeerManagerFixture(knownPeers: Seq[InetSocketAddress] = Seq.empty) extends AkkaFixture { + val dir: File = createTempDir + + val settings: ErgoSettings = { + val base = initSettings.copy(directory = dir.getAbsolutePath) + base.copy( + scorexSettings = base.scorexSettings.copy( + network = base.scorexSettings.network.copy( + knownPeers = knownPeers + ) + ) + ) + } + + val scorexContext: ScorexContext = ScorexContext(Seq.empty, None, None) + val peerManager: TestActorRef[PeerManager] = + TestActorRef(new PeerManager(settings, scorexContext)) + } + + private def withFixture(testCode: PeerManagerFixture => Any): Unit = { + val f = new PeerManagerFixture + try { + testCode(f) + } finally { + Await.result(f.system.terminate(), Duration.Inf) + } + } + + private def peerSpec(address: InetSocketAddress): PeerSpec = + defaultPeerSpec.copy(declaredAddress = Some(address)) + + private def peerInfo(address: InetSocketAddress, + lastHandshake: Long = 0L, + connectionType: Option[ConnectionDirection] = None, + lastActivity: Long = 0L): PeerInfo = + PeerInfo( + defaultPeerSpec.copy(declaredAddress = Some(address)), + lastHandshake, + connectionType, + lastActivity + ) + + private def address(i: Int): InetSocketAddress = new InetSocketAddress(s"8.8.${i / 256}.${i % 256}", 9000 + i) + + private def seenPeers(howMany: Int, + peers: Map[InetSocketAddress, PeerInfo], + blacklisted: Seq[InetAddress] = Seq.empty): Seq[PeerInfo] = + SeenPeers(howMany).choose(peers, blacklisted, ScorexContext(Seq.empty, None, None)) + + private def connectedPeer(address: InetSocketAddress): ConnectedPeer = { + val localAddress = new InetSocketAddress("127.0.0.1", 9002) + ConnectedPeer( + ConnectionId(address, localAddress, Outgoing), + ActorRef.noSender, + None + ) + } + + property("PeerManager should keep a connected peer during old-peer cleanup and not purge untried peers") { + withFixture { f => + import f._ + val address = new InetSocketAddress("8.8.8.8", 9001) + val spec = peerSpec(address) + val probe = TestProbe() + + probe.send(peerManager, AddPeerIfEmpty(spec)) + probe.send(peerManager, GetAllPeers) + val peers1 = probe.expectMsgType[Map[InetSocketAddress, PeerInfo]] + peers1.keys should contain(address) + + probe.send(peerManager, HandshakedPeer(connectedPeer(address))) + probe.send(peerManager, CleanupOldPeers) + probe.send(peerManager, GetAllPeers) + val peers2 = probe.expectMsgType[Map[InetSocketAddress, PeerInfo]] + peers2.keys should contain(address) + + probe.send(peerManager, DisconnectedPeer(connectedPeer(address))) + probe.send(peerManager, CleanupOldPeers) + probe.send(peerManager, GetAllPeers) + val peers3 = probe.expectMsgType[Map[InetSocketAddress, PeerInfo]] + // the peer was never handshaked with (lastHandshake == 0), so it is exempt + // from handshake-age cleanup + peers3.keys should contain(address) + } + } + + property("PeerManager should protect an inbound peer whose advertised address differs from the socket address") { + withFixture { f => + import f._ + // for an inbound connection the transport endpoint carries the peer's ephemeral + // source port, while the peer database is keyed by the advertised listening address + val socketAddress = new InetSocketAddress("8.8.8.8", 54321) + val advertised = new InetSocketAddress("8.8.8.8", 9001) + val oldTs = System.currentTimeMillis() - PeerDatabase.KnownPeerMaxAgeMs - 1000 + val probe = TestProbe() + + probe.send(peerManager, AddOrUpdatePeer(peerInfo(advertised, lastHandshake = oldTs))) + + val localAddress = new InetSocketAddress("127.0.0.1", 9002) + val inbound = ConnectedPeer( + ConnectionId(socketAddress, localAddress, Incoming), + ActorRef.noSender, + Some(peerInfo(advertised, lastHandshake = oldTs)) + ) + probe.send(peerManager, HandshakedPeer(inbound)) + probe.send(peerManager, CleanupOldPeers) + probe.send(peerManager, GetAllPeers) + val peers1 = probe.expectMsgType[Map[InetSocketAddress, PeerInfo]] + peers1.keys should contain(advertised) + + probe.send(peerManager, DisconnectedPeer(inbound)) + probe.send(peerManager, CleanupOldPeers) + probe.send(peerManager, GetAllPeers) + val peers2 = probe.expectMsgType[Map[InetSocketAddress, PeerInfo]] + peers2.keys should not contain advertised + } + } + + property("PeerManager should keep unavailable configured seed peers during cleanup") { + val seed = new InetSocketAddress("8.8.8.8", 9001) + val f = new PeerManagerFixture(Seq(seed)) + try { + import f._ + val probe = TestProbe() + + // the seed is added on startup with lastHandshake == 0 + probe.send(peerManager, GetAllPeers) + val peers1 = probe.expectMsgType[Map[InetSocketAddress, PeerInfo]] + peers1.keys should contain(seed) + + probe.send(peerManager, CleanupOldPeers) + probe.send(peerManager, GetAllPeers) + val peers2 = probe.expectMsgType[Map[InetSocketAddress, PeerInfo]] + peers2.keys should contain(seed) + } finally { + Await.result(f.system.terminate(), Duration.Inf) + } + } + + property("SeenPeers should return empty for non-positive or empty input") { + seenPeers(0, Map.empty) shouldBe empty + seenPeers(-1, Map.empty) shouldBe empty + seenPeers(5, Map.empty) shouldBe empty + } + + property("SeenPeers should return at most howMany peers") { + val peers = (1 to 10).map(i => address(i) -> peerInfo(address(i), lastHandshake = 1L)).toMap + val chosen = seenPeers(3, peers) + chosen.size should be <= 3 + chosen.size should be > 0 + } + + property("SeenPeers should not return peers with neither handshake nor connection type") { + val good = (1 to 5).map(i => address(i) -> peerInfo(address(i), lastHandshake = 1L)).toMap + val bad = (6 to 10).map(i => address(i) -> peerInfo(address(i))).toMap + val chosen = seenPeers(10, good ++ bad) + chosen.map(_.peerSpec.declaredAddress.get).toSet.intersect(bad.keys.toSet) shouldBe empty + chosen.size shouldBe 5 + } + + property("SeenPeers should exclude blacklisted peers") { + val peers = (1 to 10).map { i => + val addr = address(i) + addr -> peerInfo(addr, lastHandshake = 1L) + }.toMap + val blacklistedIp = InetAddress.getByName("8.8.8.1") + val chosen = (1 to 100).flatMap(_ => seenPeers(10, peers, Seq(blacklistedIp))).toSet + chosen.map(_.peerSpec.declaredAddress.get.getAddress).toSet should not contain blacklistedIp + } + + property("SeenPeers should prefer recently active peers") { + val now = System.currentTimeMillis() + val recent = (1 to 5).map(i => address(i) -> peerInfo(address(i), lastHandshake = 1L, lastActivity = now)).toMap + val old = (6 to 10).map(i => address(i) -> peerInfo(address(i), lastHandshake = 1L, lastActivity = 0L)).toMap + val chosen = seenPeers(10, recent ++ old) + chosen.size shouldBe 5 + chosen.map(_.lastStoredActivityTime).toSet should contain only now + } + + property("SeenPeers should be able to reach any peer in a small DB over multiple calls") { + val peers = (1 to 50).map(i => address(i) -> peerInfo(address(i), lastHandshake = 1L)).toMap + val returned = (1 to 200).flatMap(_ => seenPeers(5, peers)).map(_.peerSpec.declaredAddress.get).toSet + returned.size should be >= 45 + } + + property("SeenPeers should handle a large DB without materializing the full map") { + val peers = (1 to 5000).map(i => address(i) -> peerInfo(address(i), lastHandshake = 1L)).toMap + val chosen = seenPeers(8, peers) + chosen.size shouldBe 8 + chosen.toSet.size shouldBe 8 + } + + property("SeenPeers should return all peers when howMany exceeds eligible count") { + val peers = (1 to 5).map(i => address(i) -> peerInfo(address(i), lastHandshake = 1L)).toMap + seenPeers(10, peers).size shouldBe 5 + } + + property("SeenPeers may skip eligible peers outside the bounded scan window") { + val eligibleAddr = address(1) + val eligible = Map(eligibleAddr -> peerInfo(eligibleAddr, lastHandshake = 1L)) + // all other peers are ineligible (never handshaked, no connection record) + val ineligible = (2 to 2000).map(i => address(i) -> peerInfo(address(i))).toMap + val peers = eligible ++ ineligible + // With a single eligible peer among 2000, most random scan windows contain only + // ineligible records and yield an empty result; windows that do cover the eligible + // peer return only it. This is the accepted bounded-work tradeoff: the scan never + // examines more than a bounded window even though eligible peers may be missed. + val chosen = (1 to 20).flatMap(_ => seenPeers(5, peers)) + chosen.size should be <= 20 + chosen.forall(_.peerSpec.declaredAddress.contains(eligibleAddr)) shouldBe true + } + +} diff --git a/src/test/scala/org/ergoplatform/nodeView/history/PopowProcessorSpecification.scala b/src/test/scala/org/ergoplatform/nodeView/history/PopowProcessorSpecification.scala index d9ace006c3..93e113a14b 100644 --- a/src/test/scala/org/ergoplatform/nodeView/history/PopowProcessorSpecification.scala +++ b/src/test/scala/org/ergoplatform/nodeView/history/PopowProcessorSpecification.scala @@ -1,13 +1,17 @@ package org.ergoplatform.nodeView.history +import org.ergoplatform.mining.AutolykosPowScheme import org.ergoplatform.modifiers.ErgoFullBlock import org.ergoplatform.modifiers.history.popow.PoPowHeader import org.ergoplatform.nodeView.state.StateType +import org.ergoplatform.settings.NipopowSettings import org.ergoplatform.utils.ErgoCorePropertyTest +import org.ergoplatform.wallet.utils.FileUtils import scorex.util.ModifierId -class PopowProcessorSpecification extends ErgoCorePropertyTest { +class PopowProcessorSpecification extends ErgoCorePropertyTest with FileUtils { import org.ergoplatform.utils.HistoryTestHelpers._ + import org.ergoplatform.utils.ErgoNodeTestConstants.{settings => baseSettings} import org.ergoplatform.utils.generators.ChainGenerator._ private def genHistory(genesisIdOpt: Option[ModifierId], popowBootstrap: Boolean) = @@ -15,6 +19,21 @@ class PopowProcessorSpecification extends ErgoCorePropertyTest { epochLength = 10000, useLastEpochs = 3, initialDiffOpt = None, genesisIdOpt) .ensuring(_.bestFullBlockOpt.isEmpty) + private def genRealPowHistory(genesisIdOpt: Option[ModifierId], + realPowScheme: AutolykosPowScheme): ErgoHistory = { + val realPowSettings = baseSettings.copy( + directory = createTempDir.getAbsolutePath, + chainSettings = baseSettings.chainSettings.copy(powScheme = realPowScheme, genesisId = genesisIdOpt), + nodeSettings = baseSettings.nodeSettings.copy( + stateType = StateType.Utxo, + verifyTransactions = true, + blocksToKeep = -1, + nipopowSettings = NipopowSettings(nipopowBootstrap = true, p2pNipopows = 1) + ) + ) + ErgoHistory.readOrGenerate(realPowSettings)(null).ensuring(_.bestFullBlockOpt.isEmpty) + } + val toPoPoWChain = (c: Seq[ErgoFullBlock]) => c.map(b => PoPowHeader.fromBlock(b).get) property("popow proof application") { @@ -32,4 +51,22 @@ class PopowProcessorSpecification extends ErgoCorePropertyTest { receiverHistory.bestHeaderOpt.get shouldBe senderHistory.bestHeaderOpt.get } + property("popow proof application rejects headers failing real Autolykos validation") { + val senderHistory = genHistory(None, popowBootstrap = false) + val senderChain = genChain(80, senderHistory) + applyChain(senderHistory, senderChain) + + val popowProofBytes = senderHistory.popowProofBytes().get + val realPowScheme = new AutolykosPowScheme(baseSettings.chainSettings.powScheme.k, baseSettings.chainSettings.powScheme.n) + val receiverHistory = genRealPowHistory(senderHistory.bestHeaderAtHeight(1).map(_.id), realPowScheme) + val popowProof = receiverHistory.nipopowSerializer.parseBytes(popowProofBytes) + + popowProof.headersChain.exists(h => realPowScheme.validate(h).isFailure) shouldBe true + + receiverHistory.headersHeight shouldBe 0 + receiverHistory.applyPopowProof(popowProof) + receiverHistory.headersHeight shouldBe 0 + receiverHistory.bestHeaderOpt shouldBe None + } + } diff --git a/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolSpec.scala b/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolSpec.scala index 1520a9f032..d35e22703d 100644 --- a/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolSpec.scala +++ b/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolSpec.scala @@ -10,6 +10,7 @@ import org.ergoplatform.settings.{ErgoSettings, ErgoValidationSettingsUpdate, Pa import org.ergoplatform.utils.{ErgoTestHelpers, RandomWrapper} import org.scalatest.flatspec.AnyFlatSpec import org.scalatestplus.scalacheck.ScalaCheckPropertyChecks +import scorex.crypto.authds.ADKey import scorex.util.encode.Base16 import sigma.ast.ByteArrayConstant import sigma.Colls @@ -29,6 +30,13 @@ class ErgoMemPoolSpec extends AnyFlatSpec import org.ergoplatform.utils.generators.ErgoCoreTransactionGenerators._ import org.ergoplatform.utils.generators.ValidBlocksGenerators._ + private def feeTx(inputSeed: Byte, fee: Long): ErgoTransaction = { + ErgoTransaction( + IndexedSeq(new Input(ADKey @@ Array.fill(32)(inputSeed), emptyProverResult)), + IndexedSeq(new ErgoBoxCandidate(fee, feeProp, creationHeight = 0)) + ) + } + it should "accept valid transaction" in { val (us, bh) = createUtxoState(settings) val genesis = validFullBlock(None, us, bh) @@ -486,6 +494,31 @@ class ErgoMemPoolSpec extends AnyFlatSpec pool.stats.takenTxns shouldBe (family_depth + 1) * txs.size } + it should "not recommend fee below node minimal fee" in { + val feeSettings = settings.copy(nodeSettings = settings.nodeSettings.copy(minimalFeeAmount = 1000000L)) + val minimalFee = feeSettings.nodeSettings.minimalFeeAmount + val now = System.currentTimeMillis() + val lowFeeHistogram = FeeHistogramBin(nTxns = 1, totalFee = minimalFee / 2) :: + List.fill(MemPoolStatistics.nHistogramBins - 1)(FeeHistogramBin(0, 0)) + val stats = MemPoolStatistics(now, takenTxns = 1, snapTime = now, histogram = lowFeeHistogram) + val pool = new ErgoMemPool(OrderedTxPool.empty(feeSettings), stats, SortingOption.FeePerByte)(feeSettings) + + pool.getRecommendedFee(expectedWaitTimeMinutes = 0, txSize = 1024) shouldBe minimalFee + } + + it should "not let idle uptime dominate expected wait time" in { + val feeSettings = settings.copy(nodeSettings = settings.nodeSettings.copy(minimalFeeAmount = 1000000L)) + val minimalFee = feeSettings.nodeSettings.minimalFeeAmount + val poolWithHigherFeeTx = ErgoMemPool.empty(feeSettings) + .put(UnconfirmedTransaction(feeTx(inputSeed = 1, fee = minimalFee * 100), None)) + val now = System.currentTimeMillis() + val staleMeasurementStart = now - 365L * 24 * 60 * 60 * 1000 + val staleStats = MemPoolStatistics(staleMeasurementStart, takenTxns = 1, snapTime = now) + val pool = new ErgoMemPool(poolWithHigherFeeTx.pool, staleStats, SortingOption.FeePerByte)(feeSettings) + + pool.getExpectedWaitTime(txFee = minimalFee, txSize = 1024) should be <= MemPoolStatistics.measurementIntervalMsec.toLong + } + it should "put not adding transaction twice" in { val pool = ErgoMemPool.empty(settings).pool val tx = invalidErgoTransactionGen.sample.get diff --git a/src/test/scala/org/ergoplatform/nodeView/viewholder/ErgoNodeViewHolderSpec.scala b/src/test/scala/org/ergoplatform/nodeView/viewholder/ErgoNodeViewHolderSpec.scala index dfd49bc78f..2d8e8f4359 100644 --- a/src/test/scala/org/ergoplatform/nodeView/viewholder/ErgoNodeViewHolderSpec.scala +++ b/src/test/scala/org/ergoplatform/nodeView/viewholder/ErgoNodeViewHolderSpec.scala @@ -23,7 +23,7 @@ import org.ergoplatform.nodeView.mempool.ErgoMemPoolUtils.ProcessingOutcome.Acce import org.ergoplatform.wallet.utils.FileUtils import scorex.crypto.authds.{ADKey, SerializedAdProof} import scorex.util.{ModifierId, bytesToId} -import org.ergoplatform.settings.Constants.TrueTree +import org.ergoplatform.settings.Constants.{FalseTree, TrueTree} class ErgoNodeViewHolderSpec extends ErgoCorePropertyTest with NodeViewTestOps with FileUtils { import org.ergoplatform.utils.ErgoNodeTestConstants._ @@ -580,6 +580,44 @@ class ErgoNodeViewHolderSpec extends ErgoCorePropertyTest with NodeViewTestOps w } } + private val t21 = TestCase("txScriptFailure carries failing transaction id") { fixture => + import fixture._ + + val (us, bh) = createUtxoState(fixture.settings) + val wus = WrappedUtxoState(us, bh, fixture.settings) + + val genesis = validFullBlock(None, wus) + + // Apply genesis through the standard NVH route first. + applyBlock(genesis) shouldBe 'success + val wusAfterGenesis = wus.applyModifier(genesis)(_ => ()).get + + // Create a valid tx that pays to a FalseTree output. + val box = wusAfterGenesis.takeBoxes(1).head + val validTx = validTransactionFromBoxes(IndexedSeq(box), outputsProposition = FalseTree) + + val validBlock = validFullBlock(Some(genesis), wusAfterGenesis, Seq(validTx)) + + // Apply valid block and advance wrapped state. + applyBlock(validBlock) shouldBe 'success + val wusAfterValidBlock = wusAfterGenesis.applyModifier(validBlock)(_ => ()).get + + // Create a tx spending the FalseTree output; prover cannot sign it, so it has empty proofs. + val falseTreeBox = validTx.outputs.head + val invalidTx = validTransactionFromBoxes(IndexedSeq(falseTreeBox)) + + val invalidBlock = validFullBlock(Some(validBlock), wusAfterValidBlock, Seq(invalidTx)) + + subscribeEvents(classOf[SemanticallyFailedModification]) + + if (verifyTransactions) { + applyBlock(invalidBlock) shouldBe 'success + + val semFailed = expectMsgType[SemanticallyFailedModification] + ErgoNodeViewHolder.extractFailedTxId(semFailed.error) shouldBe Some(invalidTx.id) + } + } + val cases: List[TestCase] = List(t0, t1, t2, t3, t3a, t4, t5, t6, t7, t8, t9) NodeViewTestConfig.allConfigs.foreach { c => @@ -590,7 +628,7 @@ class ErgoNodeViewHolderSpec extends ErgoCorePropertyTest with NodeViewTestOps w } } - val verifyingTxCases: List[TestCase] = List(t10, t11, t12, t13, t20) + val verifyingTxCases: List[TestCase] = List(t10, t11, t12, t13, t20, t21) NodeViewTestConfig.verifyTxConfigs.foreach { c => verifyingTxCases.foreach { t => diff --git a/src/test/scala/org/ergoplatform/nodeView/wallet/ErgoWalletServiceSpec.scala b/src/test/scala/org/ergoplatform/nodeView/wallet/ErgoWalletServiceSpec.scala index cc261eebc5..284952b6f4 100644 --- a/src/test/scala/org/ergoplatform/nodeView/wallet/ErgoWalletServiceSpec.scala +++ b/src/test/scala/org/ergoplatform/nodeView/wallet/ErgoWalletServiceSpec.scala @@ -7,7 +7,7 @@ import org.ergoplatform.modifiers.mempool.{ErgoTransaction, UnconfirmedTransacti import org.ergoplatform.nodeView.mempool.ErgoMemPoolReader import org.ergoplatform.nodeView.wallet.WalletScanLogic.ScanResults import org.ergoplatform.nodeView.wallet.persistence.{OffChainRegistry, WalletRegistry, WalletStorage} -import org.ergoplatform.nodeView.wallet.requests.{AssetIssueRequest, PaymentRequest} +import org.ergoplatform.nodeView.wallet.requests.{AssetIssueRequest, BurnTokensRequest, PaymentRequest} import org.ergoplatform.nodeView.wallet.scanning.{EqualsScanningPredicate, ScanRequest, ScanWalletInteraction} import org.ergoplatform.sdk.SecretString import org.ergoplatform.sdk.wallet.secrets.{DerivationPath, ExtendedSecretKey} @@ -28,6 +28,7 @@ import scorex.db.{LDBKVStore, LDBVersionedStore} import scorex.util.encode.Base16 import sigma.Extensions.ArrayOps import sigma.ast.{ByteArrayConstant, EvaluatedValue, FalseLeaf, SType} +import sigmastate.eval.Extensions._ import sigmastate.helpers.TestingHelpers.testBox import scala.collection.compat.immutable.ArraySeq @@ -274,6 +275,101 @@ class ErgoWalletServiceSpec } } + property("asset issuance should be independent of burn request order") { + withVersionedStore(2) { versionedStore => + withStore { store => + val wState = initialState(store, versionedStore) + val existingAssetAmount = 10L + val burnAmount = 3L + val issueAmount = 7L + val inputBoxes = boxesAvailable( + makeGenesisBlock(pks.head.pubkey, Seq(newAssetIdStub -> existingAssetAmount)), + pks.head.pubkey + ) + val existingTokenId = inputBoxes.flatMap(_.additionalTokens.toArray).head._1 + val encodedBoxes = inputBoxes.map(box => Base16.encode(ErgoBoxSerializer.toBytes(box))) + val burnRequest = BurnTokensRequest(Array(existingTokenId -> burnAmount)) + val paymentRequest = PaymentRequest(pks.head, 1000000L, Array.empty, Map.empty) + val issueRequest = AssetIssueRequest( + address = pks.head, + valueOpt = Some(10000000L), + amount = issueAmount, + name = "test-name", + description = "test-description", + decimals = 4, + registers = Option.empty + ) + val boxSelector = new ReplaceCompactCollectBoxSelector( + settings.walletSettings.maxInputs, + settings.walletSettings.optimalInputs, + None + ) + + val requestOrders = Seq( + Seq(burnRequest, issueRequest), + Seq(issueRequest, burnRequest) + ) ++ Seq(burnRequest, issueRequest, paymentRequest).permutations.toSeq + + requestOrders.foreach { requests => + val result = generateUnsignedTransaction( + wState, + boxSelector, + requests, + inputsRaw = encodedBoxes, + dataInputsRaw = Seq.empty + ) + val requestOrder = requests.map(_.getClass.getSimpleName).mkString(", ") + withClue(s"request order: $requestOrder; failure: ${result.failed.map(_.getMessage).toOption}") { + result.isSuccess shouldBe true + } + + val (tx, selectedInputs, _) = result.get + val issuedTokenId = selectedInputs.head.id.toTokenId + val issueOutputs = tx.outputCandidates.filter( + _.additionalTokens.toArray.exists { case (tokenId, _) => tokenId == issuedTokenId } + ) + issueOutputs should have size 1 + issueOutputs.head.value shouldBe issueRequest.valueOpt.get + issueOutputs.head.ergoTree shouldBe pks.head.script + issueOutputs.head.additionalTokens.toArray should contain(issuedTokenId -> issueAmount) + issueOutputs.head.additionalRegisters shouldBe Map( + ErgoBox.R4 -> ByteArrayConstant("test-name".getBytes("UTF-8")), + ErgoBox.R5 -> ByteArrayConstant("test-description".getBytes("UTF-8")), + ErgoBox.R6 -> ByteArrayConstant("4".getBytes("UTF-8")) + ) + + if (requests.contains(paymentRequest)) { + val paymentOutputs = tx.outputCandidates.filter(_.value == paymentRequest.value) + paymentOutputs should have size 1 + paymentOutputs.head.ergoTree shouldBe pks.head.script + paymentOutputs.head.additionalTokens.toArray shouldBe empty + paymentOutputs.head.additionalRegisters shouldBe empty + } + + selectedInputs + .flatMap(_.additionalTokens.toArray) + .collect { case (tokenId, amount) if tokenId == issuedTokenId => amount } + .sum shouldBe 0L + tx.outputCandidates + .flatMap(_.additionalTokens.toArray) + .collect { case (tokenId, amount) if tokenId == issuedTokenId => amount } + .sum shouldBe issueAmount + + val selectedExistingAmount = selectedInputs + .flatMap(_.additionalTokens.toArray) + .collect { case (tokenId, amount) if tokenId == existingTokenId => amount } + .sum + val outputExistingAmount = tx.outputCandidates + .flatMap(_.additionalTokens.toArray) + .collect { case (tokenId, amount) if tokenId == existingTokenId => amount } + .sum + selectedExistingAmount - outputExistingAmount shouldBe burnAmount + selectedInputs.map(_.value).sum shouldBe tx.outputCandidates.map(_.value).sum + } + } + } + } + property("it should process unlock using preEip3Derivation") { withVersionedStore(2) { versionedStore => withStore { store => diff --git a/src/test/scala/org/ergoplatform/utils/generators/ErgoNodeGenerators.scala b/src/test/scala/org/ergoplatform/utils/generators/ErgoNodeGenerators.scala index b4043fe3e7..c4b03e7072 100644 --- a/src/test/scala/org/ergoplatform/utils/generators/ErgoNodeGenerators.scala +++ b/src/test/scala/org/ergoplatform/utils/generators/ErgoNodeGenerators.scala @@ -24,7 +24,7 @@ object ErgoNodeGenerators { } yield { val chain = genHeaderChain(m * mulM + k, diffBitsOpt = None, useRealTs = false) val popowChain = popowHeaderChain(chain) - val params = PoPowParams(m, k, continuous = false) + val params = PoPowParams(m, k, continuous = false).get nipopowAlgos.prove(popowChain)(params).get } } diff --git a/src/test/scala/scorex/core/network/NetworkControllerSpec.scala b/src/test/scala/scorex/core/network/NetworkControllerSpec.scala index 8bc287d6d9..1265b535ab 100644 --- a/src/test/scala/scorex/core/network/NetworkControllerSpec.scala +++ b/src/test/scala/scorex/core/network/NetworkControllerSpec.scala @@ -3,6 +3,7 @@ package scorex.core.network import akka.actor.ActorRef import akka.io.Tcp import akka.testkit.{TestActorRef, TestProbe} +import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages.DisconnectedPeer import org.ergoplatform.network.message.MessageConstants.MessageCode import org.ergoplatform.network.peer.PeerInfo import org.ergoplatform.utils.ErgoCorePropertyTest @@ -25,6 +26,8 @@ class NetworkControllerSpec extends ErgoCorePropertyTest { val scorexContext: ScorexContext = ScorexContext(Seq.empty, None, None) + case class EstablishedConnection(connectionProbe: TestProbe, handlerRef: ActorRef) + def createController(maxConnections: Int): (TestActorRef[NetworkController], TestProbe, TestProbe) = { val peerManagerProbe = TestProbe("PeerManager") val tcpManagerProbe = TestProbe("TcpManager") @@ -56,6 +59,33 @@ class NetworkControllerSpec extends ErgoCorePropertyTest { peerManagerProbe: TestProbe, remoteAddress: InetSocketAddress ): InetSocketAddress = { + beginIncomingConnection(controller, peerManagerProbe, remoteAddress) + remoteAddress + } + + def establishIncomingConnectionWithHandler( + controller: TestActorRef[NetworkController], + peerManagerProbe: TestProbe, + remoteAddress: InetSocketAddress + ): EstablishedConnection = { + val connectionProbe = beginIncomingConnection( + controller, + peerManagerProbe, + remoteAddress + ) + + val handlerRef = connectionProbe.expectMsgType[Tcp.Register].handler + connectionProbe.expectMsg(Tcp.ResumeReading) + connectionProbe.expectMsgType[Tcp.Write] + + EstablishedConnection(connectionProbe, handlerRef) + } + + private def beginIncomingConnection( + controller: TestActorRef[NetworkController], + peerManagerProbe: TestProbe, + remoteAddress: InetSocketAddress + ): TestProbe = { val localAddress = settings.scorexSettings.network.bindAddress val connectionProbe = TestProbe("Connection") @@ -66,7 +96,7 @@ class NetworkControllerSpec extends ErgoCorePropertyTest { controller ! ConnectionConfirmed(ConnectionId(remoteAddress, localAddress, Incoming), handlerRef) } - remoteAddress + connectionProbe } def establishOutgoingConnection( @@ -191,6 +221,97 @@ class NetworkControllerSpec extends ErgoCorePropertyTest { } } + property("blacklisting should close exactly the live connections for the banned IP") { + withFixture { f => + implicit val system = f.system + val (controller, peerManagerProbe, _) = f.createController(maxConnections = 30) + val disconnectProbe = TestProbe("DisconnectedPeers") + f.system.eventStream.subscribe(disconnectProbe.ref, classOf[DisconnectedPeer]) + + val firstAddress = new InetSocketAddress("192.0.2.10", 9101) + val secondAddress = new InetSocketAddress("192.0.2.10", 9102) + val unrelatedAddress = new InetSocketAddress("198.51.100.20", 9201) + val first = f.establishIncomingConnectionWithHandler( + controller, + peerManagerProbe, + firstAddress + ) + val second = f.establishIncomingConnectionWithHandler( + controller, + peerManagerProbe, + secondAddress + ) + val unrelated = f.establishIncomingConnectionWithHandler( + controller, + peerManagerProbe, + unrelatedAddress + ) + + peerManagerProbe.send(controller, Blacklisted(firstAddress)) + + first.connectionProbe.expectMsg(Tcp.Abort) + second.connectionProbe.expectMsg(Tcp.Abort) + unrelated.connectionProbe.expectNoMessage(200.millis) + + val duplicateBeforeTermination = TestProbe("DuplicateBeforeTermination") + duplicateBeforeTermination.send( + controller, + Tcp.Connected(secondAddress, settings.scorexSettings.network.bindAddress) + ) + duplicateBeforeTermination.expectMsg(Tcp.Close) + + first.connectionProbe.watch(first.handlerRef) + second.connectionProbe.watch(second.handlerRef) + first.connectionProbe.send(first.handlerRef, Tcp.Aborted) + second.connectionProbe.send(second.handlerRef, Tcp.Aborted) + first.connectionProbe.expectTerminated(first.handlerRef) + second.connectionProbe.expectTerminated(second.handlerRef) + + val disconnectedAddresses = disconnectProbe.receiveN(2, 2.seconds).collect { + case DisconnectedPeer(peer) => peer.connectionId.remoteAddress + }.toSet + disconnectedAddresses shouldBe Set(firstAddress, secondAddress) + + val replacement = TestProbe("ReplacementConnection") + replacement.send( + controller, + Tcp.Connected(secondAddress, settings.scorexSettings.network.bindAddress) + ) + peerManagerProbe.expectMsgPF(1.second) { + case ConfirmConnection(connectionId, connectionRef) => + connectionId.remoteAddress shouldBe secondAddress + connectionRef shouldBe replacement.ref + } + + val unrelatedDuplicate = TestProbe("UnrelatedDuplicate") + unrelatedDuplicate.send( + controller, + Tcp.Connected(unrelatedAddress, settings.scorexSettings.network.bindAddress) + ) + unrelatedDuplicate.expectMsg(Tcp.Close) + } + } + + property("blacklisting should match by IP when the exact socket is absent") { + withFixture { f => + val (controller, peerManagerProbe, _) = f.createController(maxConnections = 30) + val siblingAddress = new InetSocketAddress("192.0.2.30", 9301) + val missingSocketAddress = new InetSocketAddress("192.0.2.30", 9399) + val sibling = f.establishIncomingConnectionWithHandler( + controller, + peerManagerProbe, + siblingAddress + ) + + peerManagerProbe.send(controller, Blacklisted(missingSocketAddress)) + + sibling.connectionProbe.expectMsg(Tcp.Abort) + sibling.connectionProbe.watch(sibling.handlerRef) + sibling.connectionProbe.send(sibling.handlerRef, Tcp.Aborted) + sibling.connectionProbe.expectTerminated(sibling.handlerRef) + } + } + property("outgoing connection should be accepted when total below maxConnections") { withFixture { f => val (controller, peerManagerProbe, tcpManagerProbe) = f.createController(maxConnections = 10) diff --git a/src/test/scala/scorex/core/network/PeerConnectionHandlerSpecification.scala b/src/test/scala/scorex/core/network/PeerConnectionHandlerSpecification.scala new file mode 100644 index 0000000000..f8319787f4 --- /dev/null +++ b/src/test/scala/scorex/core/network/PeerConnectionHandlerSpecification.scala @@ -0,0 +1,167 @@ +package scorex.core.network + +import akka.io.Tcp +import akka.testkit.{TestActorRef, TestProbe} +import akka.util.ByteString +import org.ergoplatform.network.message.{ + GetPeersSpec, + Message, + MessageSpec, + UtxoSnapshotChunkSpec +} +import org.ergoplatform.network.{Handshake, HandshakeSerializer} +import org.ergoplatform.utils.ErgoCorePropertyTest +import org.ergoplatform.utils.ErgoNodeTestConstants.{defaultPeerSpec, settings} +import scorex.core.app.ScorexContext +import scorex.testkit.utils.AkkaFixture + +import java.net.InetSocketAddress +import scala.concurrent.Await +import scala.concurrent.duration.{Duration, DurationInt} + +class PeerConnectionHandlerSpecification extends ErgoCorePropertyTest { + private final class ConnectedHandler(val connection: TestProbe, + val watcher: TestProbe, + val handler: TestActorRef[PeerConnectionHandler]) + + private def withConnectedHandler( + messageSpecs: Seq[MessageSpec[_]], + localPort: Int + )(test: ConnectedHandler => Unit): Unit = { + val fixture = new AkkaFixture + try { + implicit val system = fixture.system + implicit val ec = system.dispatcher + val connection = TestProbe("connection") + val controller = TestProbe("controller") + val localAddress = new InetSocketAddress("127.0.0.1", localPort) + val remoteAddress = new InetSocketAddress("127.0.0.1", localPort + 1) + val description = ConnectionDescription( + connection.ref, + ConnectionId(remoteAddress, localAddress, Incoming), + Some(localAddress), + Seq.empty + ) + val handler = TestActorRef(new PeerConnectionHandler( + settings.scorexSettings, + controller.ref, + ScorexContext(messageSpecs, None, None), + description + )) + + connection.expectMsgType[Tcp.Register] + connection.expectMsg(Tcp.ResumeReading) + connection.expectMsgType[Tcp.Write] + + val handshake = HandshakeSerializer.toBytes( + Handshake(defaultPeerSpec, System.currentTimeMillis()) + ) + connection.send(handler, Tcp.Received(ByteString(handshake))) + controller.expectMsgType[NetworkController.ReceivableMessages.Handshaked] + connection.expectMsg(Tcp.ResumeReading) + controller.watch(handler) + + test(new ConnectedHandler(connection, controller, handler)) + } finally { + Await.result(fixture.system.terminate(), Duration.Inf) + } + } + + property("abort before a fifth maximum snapshot frame is retained") { + withConnectedHandler(Seq(UtxoSnapshotChunkSpec), localPort = 9083) { fixture => + val chunkMessage = Message( + UtxoSnapshotChunkSpec, + Right(Array.fill[Byte](3999996)(1)), + None + ) + fixture.handler ! chunkMessage + val failedWrite = fixture.connection.expectMsgType[Tcp.Write] + failedWrite.data.length shouldEqual 4000013 + fixture.connection.send(fixture.handler, Tcp.CommandFailed(failedWrite)) + fixture.connection.expectMsg(Tcp.ResumeWriting) + + (2 to 4).foreach { id => + val write = Tcp.Write( + failedWrite.data, + PeerConnectionHandler.ReceivableMessages.Ack(id) + ) + fixture.connection.send(fixture.handler, Tcp.CommandFailed(write)) + fixture.connection.expectMsg(Tcp.ResumeWriting) + } + fixture.connection.expectNoMessage(200.millis) + + val overLimitWrite = Tcp.Write( + failedWrite.data, + PeerConnectionHandler.ReceivableMessages.Ack(5) + ) + fixture.connection.send( + fixture.handler, + Tcp.CommandFailed(overLimitWrite) + ) + fixture.connection.expectMsg(Tcp.ResumeWriting) + fixture.connection.expectMsg(1.second, Tcp.Abort) + fixture.watcher.expectTerminated(fixture.handler) + } + } + + property("abort before more than 64 outbound messages are buffered") { + withConnectedHandler(Seq(GetPeersSpec), localPort = 9093) { fixture => + val getPeersMessage = Message(GetPeersSpec, Right(()), None) + fixture.handler ! getPeersMessage + val failedWrite = fixture.connection.expectMsgType[Tcp.Write] + fixture.connection.send(fixture.handler, Tcp.CommandFailed(failedWrite)) + fixture.connection.expectMsg(Tcp.ResumeWriting) + + (1 until PeerConnectionHandler.MaxBufferedOutboundMessages) + .foreach(_ => fixture.handler ! getPeersMessage) + fixture.connection.expectNoMessage(200.millis) + + fixture.handler ! getPeersMessage + fixture.connection.expectMsg(1.second, Tcp.Abort) + fixture.watcher.expectTerminated(fixture.handler) + } + } + + property("account retried and acknowledged writes exactly") { + withConnectedHandler( + Seq(UtxoSnapshotChunkSpec), + localPort = 9103 + ) { fixture => + val chunkMessage = Message( + UtxoSnapshotChunkSpec, + Right(Array.fill[Byte](3999996)(1)), + None + ) + fixture.handler ! chunkMessage + val failedWrite = fixture.connection.expectMsgType[Tcp.Write] + failedWrite.data.length shouldEqual 4000013 + failedWrite.ack shouldEqual PeerConnectionHandler.ReceivableMessages.Ack(1) + + fixture.connection.send(fixture.handler, Tcp.CommandFailed(failedWrite)) + fixture.connection.expectMsg(Tcp.ResumeWriting) + + (2 to 4).foreach(_ => fixture.handler ! chunkMessage) + fixture.connection.expectNoMessage(200.millis) + + fixture.connection.send(fixture.handler, Tcp.WritingResumed) + val retriedWrite = fixture.connection.expectMsgType[Tcp.Write] + retriedWrite.data shouldEqual failedWrite.data + retriedWrite.ack shouldEqual PeerConnectionHandler.ReceivableMessages.Ack(1) + fixture.connection.send(fixture.handler, Tcp.CommandFailed(retriedWrite)) + fixture.connection.expectMsg(Tcp.ResumeWriting) + fixture.connection.expectNoMessage(200.millis) + + fixture.connection.send(fixture.handler, Tcp.WritingResumed) + val finalRetry = fixture.connection.expectMsgType[Tcp.Write] + finalRetry.data shouldEqual failedWrite.data + finalRetry.ack shouldEqual PeerConnectionHandler.ReceivableMessages.Ack(1) + fixture.connection.send( + fixture.handler, + PeerConnectionHandler.ReceivableMessages.Ack(1) + ) + + val nextWrite = fixture.connection.expectMsgType[Tcp.Write] + nextWrite.ack shouldEqual PeerConnectionHandler.ReceivableMessages.Ack(2) + } + } +}