Fix stripping stream reference in gun_http

An invalid stream reference (the websocket tuple wrapper)
was sent in the gun_data message.

Also moves autobahn to its own test suite.

5 files changed, 201 insertions, 137 deletions

diff --git a/src/gun_http.erl b/src/gun_http.erl
index 81310bf..57f72c7 100644
--- a/src/gun_http.erl
+++ b/src/gun_http.erl
@@ -265,7 +265,7 @@ handle_head(Data, State=#http_state{socket=Socket, version=ClientVersion,
 					case IsFin of
 						fin -> undefined;
 						nofin ->
-							gun_content_handler:init(ReplyTo, StreamRef,
+							gun_content_handler:init(ReplyTo, stream_ref(StreamRef),
 								Status, Headers, Handlers0)
 					end
 			end,
diff --git a/test/handlers/ws_reject_h.erl b/test/handlers/ws_reject_h.erl
new file mode 100644
index 0000000..c58e672
--- /dev/null
+++ b/test/handlers/ws_reject_h.erl
@@ -0,0 +1,7 @@
+%% This handler rejects all Websocket connections.
+-module(ws_reject_h).
+
+-export([init/2]).
+
+init(Req0, Env) ->
+    {ok, cowboy_req:reply(400, #{}, <<"Upgrade rejected">>, Req0), Env}.
diff --git a/test/ws_SUITE.erl b/test/ws_SUITE.erl
index b6e10d2..3ad220c 100644
--- a/test/ws_SUITE.erl
+++ b/test/ws_SUITE.erl
@@ -1,4 +1,4 @@
-%% Copyright (c) 2015-2018, Loïc Hoguin <[email protected]>
+%% Copyright (c) 2018, Loïc Hoguin <[email protected]>
 %%
 %% Permission to use, copy, modify, and/or distribute this software for any
 %% purpose with or without fee is hereby granted, provided that the above
@@ -17,152 +17,39 @@
 -compile(nowarn_export_all).
 
 -import(ct_helper, [config/2]).
+-import(ct_helper, [doc/1]).
 
 %% ct.
 
 all() ->
-	[{group, autobahn}].
+	[{group, ws}].
 
 groups() ->
-	[{autobahn, [], [autobahn_fuzzingserver]}].
+	[{ws, [], ct_helper:all(?MODULE)}].
 
-init_per_group(autobahn, Config) ->
-	%% Some systems have it named pip2.
-	Out = os:cmd("pip show autobahntestsuite ; pip2 show autobahntestsuite"),
-	case string:str(Out, "autobahntestsuite") of
-		0 ->
-			ct:pal("Skipping the autobahn group because the "
-				"Autobahn Test Suite is not installed.~nTo install it, "
-				"please follow the instructions on this page:~n~n    "
-				"http://autobahn.ws/testsuite/installation.html"),
-			{skip, "Autobahn Test Suite not installed."};
-		_ ->
-			Config
-	end.
+init_per_suite(Config) ->
+	{ok, _} = cowboy:start_clear(ws, [], #{env => #{
+		dispatch => cowboy_router:compile([{'_', [{"/", ws_reject_h, []}]}])
+	}}),
+	Port = ranch:get_port(ws),
+	[{port, Port}|Config].
 
-end_per_group(_, _) ->
-	ok.
+end_per_suite(_) ->
+	cowboy:stop_listener(ws).
 
 %% Tests.
 
-autobahn_fuzzingserver(Config) ->
-	Self = self(),
-	spawn_link(fun() -> start_port(Config, Self) end),
-	receive autobahn_ready -> ok end,
-	N = get_case_count(),
-	run_cases(0, N),
-	Report = config(priv_dir, Config) ++ "reports/clients/index.html",
-	ct:log("<h2><a href=\"~s\">Full report</a></h2>~n", [Report]),
-	ct:print("Autobahn Test Suite report: file://~s~n", [Report]),
-	log_output(),
-	terminate(),
-	{ok, HTML} = file:read_file(Report),
-	case length(binary:matches(HTML, <<"case_failed">>)) > 2 of
-		true -> error(failed);
-		false -> ok
-	end.
-
-start_port(Config, Pid) ->
-	Port = open_port({spawn, "wstest -m fuzzingserver -s " ++ config(data_dir, Config) ++ "server.json"},
-		[{line, 10000}, {cd, config(priv_dir, Config)}, binary]),
-	receive_preamble(Port, Pid),
-	receive_infinity(Port).
-
-receive_preamble(Port, Pid) ->
-	receive
-		{Port, {data, {eol, Line = <<"Ok, will run", _/bits>>}}} ->
-			Pid ! autobahn_ready,
-			io:format(user, "~s~n", [Line]);
-		{Port, {data, {eol, Line}}} ->
-			io:format(user, "~s~n", [Line]),
-			receive_preamble(Port, Pid)
-	after 5000 ->
-		terminate(),
-		error(timeout)
-	end.
-
-receive_infinity(Port) ->
-	receive
-		{Port, {data, {eol, <<"Updating reports", _/bits>>}}} ->
-			receive_infinity(Port);
-		{Port, {data, {eol, Line}}} ->
-			io:format(user, "~s~n", [Line]),
-			receive_infinity(Port)
-	end.
-
-get_case_count() ->
-	{Pid, MRef, StreamRef} = connect("/getCaseCount"),
-	receive
-		{gun_ws, Pid, StreamRef, {text, N}} ->
-			close(Pid, MRef),
-			binary_to_integer(N);
-		Msg ->
-			ct:pal("Unexpected message ~p", [Msg]),
-			terminate(),
-			error(failed)
-	end.
-
-run_cases(Total, Total) ->
-	ok;
-run_cases(N, Total) ->
-	{Pid, MRef, StreamRef} = connect(["/runCase?case=", integer_to_binary(N + 1), "&agent=Gun"]),
-	loop(Pid, MRef, StreamRef),
-	update_reports(),
-	run_cases(N + 1, Total).
-
-loop(Pid, MRef, StreamRef) ->
+reject_upgrade(Config) ->
+	doc("Ensure Websocket connections can be rejected."),
+	{ok, ConnPid} = gun:open("localhost", config(port, Config)),
+	{ok, _} = gun:await_up(ConnPid),
+	StreamRef = gun:ws_upgrade(ConnPid, "/", []),
 	receive
-		{gun_ws, Pid, StreamRef, close} ->
-			gun:ws_send(Pid, close),
-			loop(Pid, MRef, StreamRef);
-		{gun_ws, Pid, StreamRef, {close, Code, _}} ->
-			gun:ws_send(Pid, {close, Code, <<>>}),
-			loop(Pid, MRef, StreamRef);
-		{gun_ws, Pid, StreamRef, Frame} ->
-			gun:ws_send(Pid, Frame),
-			loop(Pid, MRef, StreamRef);
-		{gun_down, Pid, ws, _, _, _} ->
-			close(Pid, MRef);
-		{'DOWN', MRef, process, Pid, normal} ->
-			close(Pid, MRef);
+		{gun_response, ConnPid, StreamRef, nofin, 400, _} ->
+			{ok, <<"Upgrade rejected">>} = gun:await_body(ConnPid, StreamRef, 1000),
+			gun:close(ConnPid);
 		Msg ->
-			ct:pal("Unexpected message ~p", [Msg]),
-			close(Pid, MRef)
-	end.
-
-update_reports() ->
-	{Pid, MRef, StreamRef} = connect("/updateReports?agent=Gun"),
-	receive
-		{gun_ws, Pid, StreamRef, close} ->
-			close(Pid, MRef)
-	after 5000 ->
-		error(failed)
+			error({fail, Msg})
+	after 1000 ->
+		error(timeout)
 	end.
-
-log_output() ->
-	ok.
-
-connect(Path) ->
-	{ok, Pid} = gun:open("127.0.0.1", 33080, #{retry => 0}),
-	{ok, http} = gun:await_up(Pid),
-	MRef = monitor(process, Pid),
-	StreamRef = gun:ws_upgrade(Pid, Path, [], #{compress => true}),
-	receive
-		{gun_upgrade, Pid, StreamRef, [<<"websocket">>], _} ->
-			ok;
-		Msg ->
-			ct:pal("Unexpected message ~p", [Msg]),
-			terminate(),
-			error(failed)
-	end,
-	{Pid, MRef, StreamRef}.
-
-close(Pid, MRef) ->
-	demonitor(MRef),
-	gun:close(Pid),
-	gun:flush(Pid).
-
-terminate() ->
-	Res = os:cmd("killall wstest"),
-	io:format(user, "~s", [Res]),
-	ok.
diff --git a/test/ws_autobahn_SUITE.erl b/test/ws_autobahn_SUITE.erl
new file mode 100644
index 0000000..4a57938
--- /dev/null
+++ b/test/ws_autobahn_SUITE.erl
@@ -0,0 +1,170 @@
+%% Copyright (c) 2015-2018, Loïc Hoguin <[email protected]>
+%%
+%% Permission to use, copy, modify, and/or distribute this software for any
+%% purpose with or without fee is hereby granted, provided that the above
+%% copyright notice and this permission notice appear in all copies.
+%%
+%% THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+%% WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+%% MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+%% ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+%% WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+%% ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+%% OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+
+-module(ws_autobahn_SUITE).
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-import(ct_helper, [config/2]).
+
+%% ct.
+
+all() ->
+	[{group, autobahn}].
+
+groups() ->
+	[{autobahn, [], [autobahn_fuzzingserver]}].
+
+init_per_group(autobahn, Config) ->
+	%% Some systems have it named pip2.
+	Out = os:cmd("pip show autobahntestsuite ; pip2 show autobahntestsuite"),
+	case string:str(Out, "autobahntestsuite") of
+		0 ->
+			ct:pal("Skipping the autobahn group because the "
+				"Autobahn Test Suite is not installed.~nTo install it, "
+				"please follow the instructions on this page:~n~n    "
+				"http://autobahn.ws/testsuite/installation.html"),
+			{skip, "Autobahn Test Suite not installed."};
+		_ ->
+			Config
+	end;
+init_per_group(_, _) ->
+	ok.
+
+end_per_group(_, _) ->
+	ok.
+
+%% Tests.
+
+autobahn_fuzzingserver(Config) ->
+	Self = self(),
+	spawn_link(fun() -> start_port(Config, Self) end),
+	receive autobahn_ready -> ok end,
+	N = get_case_count(),
+	run_cases(0, N),
+	Report = config(priv_dir, Config) ++ "reports/clients/index.html",
+	ct:log("<h2><a href=\"~s\">Full report</a></h2>~n", [Report]),
+	ct:print("Autobahn Test Suite report: file://~s~n", [Report]),
+	log_output(),
+	terminate(),
+	{ok, HTML} = file:read_file(Report),
+	case length(binary:matches(HTML, <<"case_failed">>)) > 2 of
+		true -> error(failed);
+		false -> ok
+	end.
+
+start_port(Config, Pid) ->
+	Port = open_port({spawn, "wstest -m fuzzingserver -s " ++ config(data_dir, Config) ++ "server.json"},
+		[{line, 10000}, {cd, config(priv_dir, Config)}, binary]),
+	receive_preamble(Port, Pid),
+	receive_infinity(Port).
+
+receive_preamble(Port, Pid) ->
+	receive
+		{Port, {data, {eol, Line = <<"Ok, will run", _/bits>>}}} ->
+			Pid ! autobahn_ready,
+			io:format(user, "~s~n", [Line]);
+		{Port, {data, {eol, Line}}} ->
+			io:format(user, "~s~n", [Line]),
+			receive_preamble(Port, Pid)
+	after 5000 ->
+		terminate(),
+		error(timeout)
+	end.
+
+receive_infinity(Port) ->
+	receive
+		{Port, {data, {eol, <<"Updating reports", _/bits>>}}} ->
+			receive_infinity(Port);
+		{Port, {data, {eol, Line}}} ->
+			io:format(user, "~s~n", [Line]),
+			receive_infinity(Port)
+	end.
+
+get_case_count() ->
+	{Pid, MRef, StreamRef} = connect("/getCaseCount"),
+	receive
+		{gun_ws, Pid, StreamRef, {text, N}} ->
+			close(Pid, MRef),
+			binary_to_integer(N);
+		Msg ->
+			ct:pal("Unexpected message ~p", [Msg]),
+			terminate(),
+			error(failed)
+	end.
+
+run_cases(Total, Total) ->
+	ok;
+run_cases(N, Total) ->
+	{Pid, MRef, StreamRef} = connect(["/runCase?case=", integer_to_binary(N + 1), "&agent=Gun"]),
+	loop(Pid, MRef, StreamRef),
+	update_reports(),
+	run_cases(N + 1, Total).
+
+loop(Pid, MRef, StreamRef) ->
+	receive
+		{gun_ws, Pid, StreamRef, close} ->
+			gun:ws_send(Pid, close),
+			loop(Pid, MRef, StreamRef);
+		{gun_ws, Pid, StreamRef, {close, Code, _}} ->
+			gun:ws_send(Pid, {close, Code, <<>>}),
+			loop(Pid, MRef, StreamRef);
+		{gun_ws, Pid, StreamRef, Frame} ->
+			gun:ws_send(Pid, Frame),
+			loop(Pid, MRef, StreamRef);
+		{gun_down, Pid, ws, _, _, _} ->
+			close(Pid, MRef);
+		{'DOWN', MRef, process, Pid, normal} ->
+			close(Pid, MRef);
+		Msg ->
+			ct:pal("Unexpected message ~p", [Msg]),
+			close(Pid, MRef)
+	end.
+
+update_reports() ->
+	{Pid, MRef, StreamRef} = connect("/updateReports?agent=Gun"),
+	receive
+		{gun_ws, Pid, StreamRef, close} ->
+			close(Pid, MRef)
+	after 5000 ->
+		error(failed)
+	end.
+
+log_output() ->
+	ok.
+
+connect(Path) ->
+	{ok, Pid} = gun:open("127.0.0.1", 33080, #{retry => 0}),
+	{ok, http} = gun:await_up(Pid),
+	MRef = monitor(process, Pid),
+	StreamRef = gun:ws_upgrade(Pid, Path, [], #{compress => true}),
+	receive
+		{gun_upgrade, Pid, StreamRef, [<<"websocket">>], _} ->
+			ok;
+		Msg ->
+			ct:pal("Unexpected message ~p", [Msg]),
+			terminate(),
+			error(failed)
+	end,
+	{Pid, MRef, StreamRef}.
+
+close(Pid, MRef) ->
+	demonitor(MRef),
+	gun:close(Pid),
+	gun:flush(Pid).
+
+terminate() ->
+	Res = os:cmd("killall wstest"),
+	io:format(user, "~s", [Res]),
+	ok.
diff --git a/test/ws_SUITE_data/server.json b/test/ws_autobahn_SUITE_data/server.json
index 902b9b3..902b9b3 100644
--- a/test/ws_SUITE_data/server.json
+++ b/test/ws_autobahn_SUITE_data/server.json