Can you post a minimal reproducer?
On Fri, Oct 7, 2016 at 9:02 PM, 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> 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(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.
>
--
Cheers,
√
--
>>>>>>>>>> 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.
