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 [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.
