Websockets client and illegal headers

[johanandren]

Moving it over from akka/akka#23026 reported by @disblader

Hello,

It seems that illegal headers crash the websocket client flow.

I am trying to connect to a websocket server which sends a server header containing a semicolon (e.g. "server: something; something"). When attempting to connect, akka seems to die. The log:

[DEBUG] [05/24/2017 15:35:39.672] [run-main-0] [EventStream(akka://default)] logger log1-Logging$DefaultLogger started
[DEBUG] [05/24/2017 15:35:39.673] [run-main-0] [EventStream(akka://default)] Default Loggers started
[DEBUG] [05/24/2017 15:35:39.785] [run-main-0] [AkkaSSLConfig(akka://default)] Initializing AkkaSSLConfig extension...
[DEBUG] [05/24/2017 15:35:39.787] [run-main-0] [AkkaSSLConfig(akka://default)] buildHostnameVerifier: created hostname verifier: com.typesafe.sslconfig.ssl.DefaultHostnameVerifier@3cf8d47b
[DEBUG] [05/24/2017 15:35:40.870] [default-akka.actor.default-dispatcher-2] [akka://default/system/IO-TCP/selectors/$a/0] Resolving localhost before connecting
[DEBUG] [05/24/2017 15:35:40.890] [default-akka.actor.default-dispatcher-4] [akka://default/system/IO-DNS] Resolution request for localhost from Actor[akka://default/system/IO-TCP/selectors/$a/0#-1706583489]
[DEBUG] [05/24/2017 15:35:40.904] [default-akka.actor.default-dispatcher-2] [akka://default/system/IO-TCP/selectors/$a/0] Attempting connection to [localhost/127.0.0.1:8080]
[DEBUG] [05/24/2017 15:35:40.909] [default-akka.actor.default-dispatcher-2] [akka://default/system/IO-TCP/selectors/$a/0] Connection established to [localhost:8080]
[ERROR] [05/24/2017 15:35:40.955] [default-akka.actor.default-dispatcher-4] [akka://default/user/StreamSupervisor-0/flow-0-0-unknown-operation] Error in stage [UpgradeStage]: unexpected element of type class akka.http.impl.engine.parsing.ParserOutput$MessageStartError
java.lang.IllegalStateException: unexpected element of type class akka.http.impl.engine.parsing.ParserOutput$MessageStartError
    at akka.http.impl.engine.ws.WebSocketClientBlueprint$UpgradeStage$1$$anon$1.onPush(WebSocketClientBlueprint.scala:107)
    at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:747)
    at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:649)
    at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:471)
    at akka.stream.impl.fusing.GraphInterpreterShell.receive(ActorGraphInterpreter.scala:423)
    at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$processEvent(ActorGraphInterpreter.scala:603)
    at akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:618)
    at akka.actor.Actor.aroundReceive(Actor.scala:497)
    at akka.actor.Actor.aroundReceive$(Actor.scala:495)
    at akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:529)
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)
    at akka.actor.ActorCell.invoke(ActorCell.scala:495)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
    at akka.dispatch.Mailbox.run(Mailbox.scala:224)
    at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
[DEBUG] [05/24/2017 15:35:40.958] [default-akka.actor.default-dispatcher-4] [akka://default/user/StreamSupervisor-0/flow-0-0-unknown-operation] Aborting tcp connection to localhost:8080 because of upstream failure: unexpected element of type class akka.http.impl.engine.parsing.ParserOutput$MessageStartError
akka.http.impl.engine.ws.WebSocketClientBlueprint$UpgradeStage$1$$anon$1.onPush(WebSocketClientBlueprint.scala:107)
akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:747)
akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:649)
akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:471)
akka.stream.impl.fusing.GraphInterpreterShell.receive(ActorGraphInterpreter.scala:423)
akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$processEvent(ActorGraphInterpreter.scala:603)
akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:618)
akka.actor.Actor.aroundReceive(Actor.scala:497)
akka.actor.Actor.aroundReceive$(Actor.scala:495)
akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:529)
akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)
akka.actor.ActorCell.invoke(ActorCell.scala:495)
akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
akka.dispatch.Mailbox.run(Mailbox.scala:224)
akka.dispatch.Mailbox.exec(Mailbox.scala:234)
java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

To reproduce, run { echo -ne "HTTP/1.0 200 OK\r\nServer: something; something\r\nContent-Length: 10\r\n\r\n1111111111"; } | nc -l 8080, and then run the following code

import akka.actor.ActorSystem
import akka.{Done, NotUsed}
import akka.http.scaladsl.Http
import akka.stream.ActorMaterializer
import akka.stream.scaladsl._
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.ws._

import scala.concurrent.Future

object HelloWorld {
  def main(args: Array[String]): Unit = {

    implicit val system = ActorSystem()
    implicit val materializer = ActorMaterializer()

    val printSink: Sink[Message, Future[Done]] = Sink.ignore
    val helloSource: Source[Message, NotUsed] = Source.single(TextMessage("hello world!"))
    val flow: Flow[Message, Message, Future[Done]] = Flow.fromSinkAndSourceMat(printSink, helloSource)(Keep.left)
    val (upgradeResponse, closed) = Http().singleWebSocketRequest(WebSocketRequest("ws://localhost:8080"), flow)

  }
}

I also tried to change akka.http.parsing.illegal-response-header-value-processing-mode to try and circumvent this issue. It does not seem to have worked, however. I can see that it works when making regular HTTP requests, but it seems to have no effect on the websocket requests. Is there anything else that can be done?

[jrudolph]

Thanks for the report, @disblader. I can confirm that it is like you report.

The reason is that in this line, the default onIllegalHeader handler is used which just throws an exception instead of using any of the configured alternatives. Should be simple enough to fix, it should use the same handler as used in the OutgoingConnectionBlueprint instead.

[disblader]

Pardon my ignorance, but what's the process for having bugs like this fixed? Is someone from the community expected to pick this up and fix it? Should I try to do it myself and then do a pull request?

[jlprat]

As it is labelled with the community tag, the core team might not be able to find some time to work on it. So if you, or any other one who is interested on this feature, have time to work on this, your contribution is highly welcome :)

Regarding how to contribute to Akka HTTP, you can go over the CONTRIBUTING file where the procedure is explained.

[disblader]

Thanks! I'll get on it straight away.

[disblader]

I think I got it, but I've got a couple of questions.

First, I'm not sure how I should handle an unexpected value here. Is there an exception that implementors generally use for these kinds of situations?

Secondly, I'm pretty bad with scalatest and I ended up in this situation where I have identical tests with different setups and I'm not sure if that's fine or not?

[jlprat]

I think it's better to discuss those on the PR directly.

[jlprat]

What I mean, is that you can submit the PR (you can name it WIP if you want) and add this comments there.

[disblader]

Fair enough. I shall do so. Thanks!

[joan38]

I'm having this issue with the Kubernetes API:

java.lang.IllegalStateException: unexpected element of type class akka.http.impl.engine.parsing.ParserOutput$MessageStartError
	at akka.http.impl.engine.ws.WebSocketClientBlueprint$UpgradeStage$1$$anon$1.onPush(WebSocketClientBlueprint.scala:107)
	at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:747)
	at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:649)
	at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:471)
	at akka.stream.impl.fusing.GraphInterpreterShell.receive(ActorGraphInterpreter.scala:410)
	at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$processEvent(ActorGraphInterpreter.scala:603)
	at akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:618)
	at akka.actor.Actor.aroundReceive(Actor.scala:497)
	at akka.actor.Actor.aroundReceive$(Actor.scala:495)
	at akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:529)
	at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)
	at akka.actor.ActorCell.invoke(ActorCell.scala:495)
	at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
	at akka.dispatch.Mailbox.run(Mailbox.scala:224)
	at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
	at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
	at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
	at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
	at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

[asm123]

I am facing the same exception. Was anybody able to resolve it?

[disblader]

I've made a fix for it, it's currently sitting in #1175, there's a couple of final touches needed and it'll probably be merged in.

As for a work-around I could not figure one out myself.