I wrote a simple test case using a graph rather than using src.runForEach.
"CipherStage in graph" should "work" in {
val clearText = "0123456789abcdef"
val clearSource = Source.single(ByteString(clearText))
val encryptedOut = Sink.head[ByteString]
val encodedKey: String = "KCl02Tjzsid09VnDl6CDpDlnm4G4VUJr8l6PNg+MHkQ="
val decodedKey = Base64.getDecoder.decode(encodedKey)
val key = new SecretKeySpec(decodedKey, 0, decodedKey.length, "AES")
val encodedIv: String = "AAAAAAAAAAAAAAAAAAAAAA=="
val decodedIv = Base64.getDecoder.decode(encodedIv)
val iv = new IvParameterSpec(decodedIv)
val graph = GraphDSL.create(encryptedOut) { implicit builder => encOut =>
import GraphDSL.Implicits._
val in: Source[ByteString, Any] = clearSource
val encryptor = new CipherStage(key, iv, Cipher.ENCRYPT_MODE)
in ~> encryptor ~> encOut
ClosedShape
}
val rg = RunnableGraph.fromGraph[Future[ByteString]](graph)
implicit val system = ActorSystem("test")
implicit val materializer = ActorMaterializer()
val blkOutFuture = rg.run()
implicit val ec = system.dispatcher
whenReady(blkOutFuture) { blkOut =>
blkOut should be (encrypt(key, iv, clearText.getBytes))
}
}
This test fails. However, when I run this test case, I get the following stack
trace:
[ERROR] [10/07/2016 12:40:03.541] [test-akka.actor.default-dispatcher-3]
[akka://test/user/StreamSupervisor-0/flow-0-0-unknown-operation] Error in stage
[com.genecloud.blockstore.CipherStage@3e77bd3f]: requirement failed: Cannot
push port (Encryptor.out) twice
java.lang.IllegalArgumentException: requirement failed: Cannot push port
(Encryptor.out) twice
at scala.Predef$.require(Predef.scala:224)
at akka.stream.stage.GraphStageLogic.push(GraphStage.scala:459)
at
com.example.blockstore.CipherStage$$anon$1$$anon$3.onUpstreamFinish(BlockstoreProcessor.scala:61)
at
akka.stream.impl.fusing.GraphInterpreter.processEvent(GraphInterpreter.scala:732)
at
akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:616)
at
akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:471)
at
akka.stream.impl.fusing.GraphInterpreterShell.init(ActorGraphInterpreter.scala:381)
at
akka.stream.impl.fusing.ActorGraphInterpreter.tryInit(ActorGraphInterpreter.scala:538)
at
akka.stream.impl.fusing.ActorGraphInterpreter.preStart(ActorGraphInterpreter.scala:586)
at akka.actor.Actor$class.aroundPreStart(Actor.scala:489)
at
akka.stream.impl.fusing.ActorGraphInterpreter.aroundPreStart(ActorGraphInterpreter.scala:529)
at akka.actor.ActorCell.create(ActorCell.scala:590)
at akka.actor.ActorCell.invokeAll$1(ActorCell.scala:461)
at akka.actor.ActorCell.systemInvoke(ActorCell.scala:483)
at akka.dispatch.Mailbox.processAllSystemMessages(Mailbox.scala:282)
at akka.dispatch.Mailbox.run(Mailbox.scala:223)
at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at
scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at
scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at
scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
I don’t see that I am pushing twice. Here is the current definition of
CipherStage:
class CipherStage(key: SecretKey, iv: IvParameterSpec, mode: Int) extends
GraphStage[FlowShape[ByteString, ByteString]] {
val in = Inlet[ByteString]("Encryptor.in")
val out = Outlet[ByteString]("Encryptor.out")
override val shape = FlowShape.of(in, out)
val log = Logger(LoggerFactory.getLogger("CipherStage"))
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
new GraphStageLogic(shape) {
val cipher: Cipher = Cipher.getInstance("AES/CBC/PKCS5Padding")
cipher.init(mode, key, iv)
setHandler(out, new OutHandler {
override def onPull(): Unit = {
pull(in)
}
})
setHandler(in, new InHandler {
override def onPush(): Unit = {
val chunk = grab(in)
log.info(s"got chunk: ${chunk}")
push(out, ByteString(cipher.update(chunk.toArray)))
}
override def onUpstreamFinish(): Unit = {
log.info(s"onUpstreamFinish")
val chunk = cipher.doFinal()
log.info(s"got final chunk: ${chunk}")
push(out, ByteString(chunk))
completeStage()
}
override def onUpstreamFailure(ex: Throwable): Unit = {
log.info(s"onUpstreamFailure")
failStage(ex)
}
})
}
}
In this graph (which is simpler than my original graph, which combined
digesting and encryption), onUpstreamFinish *is* being called, it just results
in an exception on the “push” call. I absolutely need to be able to “push”
during onStreamFinish, since with AES/CBC encryption, there are more stream
elements after the last input that must be emitted.
Anyone see what is wrong? — Eric
> On Oct 7, 2016, at 12:02, Eric Swenson <e...@swenson.org> wrote:
>
> I wrote a simple test for my CipherStage and it appears to work fine:
>
> "CipherStage" should "work" in {
> val clearText = "0123456789abcdef"
> val clearSource = Source.single(ByteString(clearText))
>
> val encodedKey: String = "KCl02Tjzsid09VnDl6CDpDlnm4G4VUJr8l6PNg+MHkQ="
> val decodedKey = Base64.getDecoder.decode(encodedKey)
> val key = new SecretKeySpec(decodedKey, 0, decodedKey.length, "AES")
>
> val encodedIv: String = "AAAAAAAAAAAAAAAAAAAAAA=="
> val decodedIv = Base64.getDecoder.decode(encodedIv)
> val iv = new IvParameterSpec(decodedIv)
>
> val src = clearSource.via(new CipherStage(key, iv, Cipher.ENCRYPT_MODE))
>
> implicit val system = ActorSystem("test")
> implicit val materializer = ActorMaterializer()
> src.runForeach(i => println(i))
> println(src)
> }
> In other words, onUpstreamFinish is called and two AES blocks are emitted (by
> the println). So it must have something to do with the graph I’m using. It
> looks like:
>
> in ~> bcast ~> dgst ~> dgstOut
> bcast ~> encryptor ~> blkOut
> Where in is a Source[ByteString,Any] (the payload), dgst is the
> DigestCalculator stage, encryptor is the CipherStage, and dgstOut and blkOut
> are the two Sink.head[ByteString] outputs of my graph. bcast is a normal
> two-output Broadcast element.
>
> Why is it that in the flow in ~> bcast ~> dgst, the onStreamFinish of dgst is
> invoked correctly, but in the flow in ~> bcast ~> encryptor, it isn’t?
>
> — Eric
>
>
>> On Oct 7, 2016, at 11:37, Eric Swenson <e...@swenson.org
>> <mailto:e...@swenson.org>> wrote:
>>
>> I have a web service which accepts an inbound payload and runs it through an
>> akka-streams pipeline that simulatenously computes the MD5 digest of the
>> payload and encrypts that payload. I’ve implemented a GraphStage that
>> performs the crypto that looks like this:
>>
>> class CipherStage(key: SecretKey, iv: IvParameterSpec, mode: Int) extends
>> GraphStage[FlowShape[ByteString, ByteString]] {
>> val in = Inlet[ByteString]("Encryptor.in")
>> val out = Outlet[ByteString]("Encryptor.out")
>> override val shape = FlowShape.of(in, out)
>>
>> override def createLogic(inheritedAttributes: Attributes): GraphStageLogic
>> = new GraphStageLogic(shape) {
>> val cipher: Cipher = Cipher.getInstance("AES/CBC/PKCS5Padding")
>> cipher.init(mode, key, iv)
>>
>> setHandler(out, new OutHandler {
>> override def onPull(): Unit = {
>> pull(in)
>> }
>> })
>>
>> setHandler(in, new InHandler {
>> override def onPush(): Unit = {
>> val chunk = grab(in)
>> emit(out, ByteString(cipher.update(chunk.toArray)))
>> pull(in)
>> }
>>
>> override def onUpstreamFinish(): Unit = {
>> val chunk = cipher.doFinal()
>> emit(out, ByteString(chunk))
>> completeStage()
>> }
>>
>> override def onUpstreamFailure(ex: Throwable): Unit = {
>> log.info <http://log.info/>(s”onUpstreamFailure: $ex")
>> failStage(ex)
>> }
>> })
>> }
>> }
>> I’m using it (and a similar digest-computing stage like this:
>>
>> def verifyAndStoreBlock(digest: String, byteStringSource: Source[ByteString,
>> Any], userUid: String, sender: ActorRef) = {
>> val digestOut = Sink.head[ByteString]
>> val blockOut = Sink.head[ByteString]
>>
>> val graph: Graph[ClosedShape, (Future[ByteString], Future[ByteString])] =
>> GraphDSL.create(digestOut, blockOut)((_,_)) { implicit builder => (dgstOut,
>> blkOut) =>
>> import GraphDSL.Implicits._
>> val in: Source[ByteString,Any] = byteStringSource
>> val bcast = builder.add(Broadcast[ByteString](2))
>> val dgst = new DigestCalculator("MD5")
>> val encryptor = new CipherStage(blockEncryptionKey,
>> blockInitializationVector, Cipher.ENCRYPT_MODE)
>>
>> in ~> bcast ~> dgst ~> dgstOut
>> bcast ~> encryptor ~> blkOut
>>
>> ClosedShape
>> }
>>
>> val rg = RunnableGraph.fromGraph[(Future[ByteString],
>> Future[ByteString])](graph)
>>
>> implicit val materializer = ActorMaterializer()
>> val (dgstOutFuture, blkOutFuture) = rg.run()
>> implicit val ec = context.dispatcher
>> for {
>> dgst <- dgstOutFuture
>> blkOut <- blkOutFuture
>> } {
>> val verify = byteStringToHexString(dgst)
>> if (verify == digest) {
>> val dataStoreActor = context.actorOf(DataStoreActor.props(dataStore))
>> dataStoreActor ! DataStoreActor.Messages.SaveBlock(digest, blkOut)
>> blockStoreLogger.logEvent(PutBlockEvent(digest, userUid))
>> sender ! Right(())
>> } else {
>> log.warning(s"Invalid digest: supplied $digest, computed: $verify")
>> blockStoreLogger.logEvent(PutBlockFailedEvent(digest, userUid, "digest
>> invalid"))
>> sender ! Left(PutStatus.DigestInvalid)
>> }
>> }
>> }
>> When I get an inbound request, the digesting works correctly, and the
>> encryption sort of works. However, onUpstreamFinish is never called (in the
>> CipherStage’s InHandler), and consequently the last AES block (with padding)
>> is not emitted correctly.
>>
>> I modelled the above CipherStage on a similar DigestCalculator stage I found
>> in the akka documentation. In the DigestCalculator graph stage, the
>> onUpstreamFinish handler is correctly called. Why it not called for the
>> CipherStage?
>>
>> — Eric
>>
>
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
