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 <[email protected]> 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 [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
