No OneTemporary
Actions

Authored By

Unknown

Size

23 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex
	index a42a9a63c..6233c3340 100644
	--- a/lib/pleroma/web/mastodon_api/websocket_handler.ex
	+++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex
	@@ -1,221 +1,257 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
	require Logger

	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.OAuth.Token
	alias Pleroma.Web.Streamer
	alias Pleroma.Web.StreamerView

	@behaviour :cowboy_websocket

	# Client ping period.
	@tick :timer.seconds(30)
	# Cowboy timeout period.
	@timeout :timer.seconds(60)
	# Hibernate every X messages
	@hibernate_every 100

	def init(%{qs: qs} = req, state) do
	with params <- Enum.into(:cow_qs.parse_qs(qs), %{}),
	sec_websocket <- :cowboy_req.header("sec-websocket-protocol", req, nil),
	access_token <- Map.get(params, "access_token"),
	{:ok, user, oauth_token} <- authenticate_request(access_token, sec_websocket),
	{:ok, topic} <- Streamer.get_topic(params["stream"], user, oauth_token, params) do
	req =
	if sec_websocket do
	:cowboy_req.set_resp_header("sec-websocket-protocol", sec_websocket, req)
	else
	req
	end

	topics =
	if topic do
	[topic]
	else
	[]
	end

	{:cowboy_websocket, req,
	%{user: user, topics: topics, oauth_token: oauth_token, count: 0, timer: nil},
	%{idle_timeout: @timeout}}
	else
	{:error, :bad_topic} ->
	Logger.debug("#{__MODULE__} bad topic #{inspect(req)}")
	req = :cowboy_req.reply(404, req)
	{:ok, req, state}

	{:error, :unauthorized} ->
	Logger.debug("#{__MODULE__} authentication error: #{inspect(req)}")
	req = :cowboy_req.reply(401, req)
	{:ok, req, state}
	end
	end

	def websocket_init(state) do
	Logger.debug(
	"#{__MODULE__} accepted websocket connection for user #{(state.user \|\| %{id: "anonymous"}).id}, topics #{state.topics}"
	)

	Enum.each(state.topics, fn topic -> Streamer.add_socket(topic, state.oauth_token) end)
	{:ok, %{state \| timer: timer()}}
	end

	# Client's Pong frame.
	def websocket_handle(:pong, state) do
	if state.timer, do: Process.cancel_timer(state.timer)
	{:ok, %{state \| timer: timer()}}
	end

	# We only receive pings for now
	def websocket_handle(:ping, state), do: {:ok, state}

	def websocket_handle({:text, text}, state) do
	with {:ok, %{} = event} <- Jason.decode(text) do
	handle_client_event(event, state)
	else
	_ ->
	Logger.error("#{__MODULE__} received non-JSON event: #{inspect(text)}")
	{:ok, state}
	end
	end

	def websocket_handle(frame, state) do
	Logger.error("#{__MODULE__} received frame: #{inspect(frame)}")
	{:ok, state}
	end

	def websocket_info({:render_with_user, view, template, item}, state) do
	user = %User{} = User.get_cached_by_ap_id(state.user.ap_id)

	unless Streamer.filtered_by_user?(user, item) do
	websocket_info({:text, view.render(template, item, user)}, %{state \| user: user})
	else
	{:ok, state}
	end
	end

	def websocket_info({:text, message}, state) do
	# If the websocket processed X messages, force an hibernate/GC.
	# We don't hibernate at every message to balance CPU usage/latency with RAM usage.
	if state.count > @hibernate_every do
	{:reply, {:text, message}, %{state \| count: 0}, :hibernate}
	else
	{:reply, {:text, message}, %{state \| count: state.count + 1}}
	end
	end

	# Ping tick. We don't re-queue a timer there, it is instead queued when :pong is received.
	# As we hibernate there, reset the count to 0.
	# If the client misses :pong, Cowboy will automatically timeout the connection after
	# `@idle_timeout`.
	def websocket_info(:tick, state) do
	{:reply, :ping, %{state \| timer: nil, count: 0}, :hibernate}
	end

	def websocket_info(:close, state) do
	{:stop, state}
	end

	# State can be `[]` only in case we terminate before switching to websocket,
	# we already log errors for these cases in `init/1`, so just do nothing here
	def terminate(_reason, _req, []), do: :ok

	def terminate(reason, _req, state) do
	Logger.debug(
	"#{__MODULE__} terminating websocket connection for user #{(state.user \|\| %{id: "anonymous"}).id}, topics #{state.topics \|\| "?"}: #{inspect(reason)}"
	)

	Enum.each(state.topics, fn topic -> Streamer.remove_socket(topic) end)
	:ok
	end

	# Public streams without authentication.
	defp authenticate_request(nil, nil) do
	{:ok, nil, nil}
	end

	# Authenticated streams.
	defp authenticate_request(access_token, sec_websocket) do
	token = access_token \|\| sec_websocket

	with true <- is_bitstring(token),
	oauth_token = %Token{user_id: user_id} <- Repo.get_by(Token, token: token),
	user = %User{} <- User.get_cached_by_id(user_id) do
	{:ok, user, oauth_token}
	else
	_ -> {:error, :unauthorized}
	end
	end

	defp timer do
	Process.send_after(self(), :tick, @tick)
	end

	defp handle_client_event(%{"type" => "subscribe", "stream" => _topic} = params, state) do
	with {_, {:ok, topic}} <-
	{:topic, Streamer.get_topic(params["stream"], state.user, state.oauth_token, params)},
	{_, false} <- {:subscribed, topic in state.topics} do
	Streamer.add_socket(topic, state.oauth_token)

	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{type: "subscribe", result: "success"})}
	], %{state \| topics: [topic \| state.topics]}}
	else
	{:subscribed, true} ->
	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{type: "subscribe", result: "ignored"})}
	], state}

	{:topic, {:error, error}} ->
	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{
	type: "subscribe",
	result: "error",
	error: error
	})}
	], state}
	end
	end

	defp handle_client_event(%{"type" => "unsubscribe", "stream" => _topic} = params, state) do
	with {_, {:ok, topic}} <-
	{:topic, Streamer.get_topic(params["stream"], state.user, state.oauth_token, params)},
	{_, true} <- {:subscribed, topic in state.topics} do
	Streamer.remove_socket(topic)

	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{type: "unsubscribe", result: "success"})}
	], %{state \| topics: List.delete(state.topics, topic)}}
	else
	{:subscribed, false} ->
	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{type: "unsubscribe", result: "ignored"})}
	], state}

	{:topic, {:error, error}} ->
	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{
	type: "unsubscribe",
	result: "error",
	error: error
	})}
	], state}
	end
	end

	+ defp handle_client_event(
	+ %{"type" => "pleroma.authenticate", "token" => access_token} = _params,
	+ state
	+ ) do
	+ with {:auth, nil, nil} <- {:auth, state.user, state.oauth_token},
	+ {:ok, user, oauth_token} <- authenticate_request(access_token, nil) do
	+ {[
	+ {:text,
	+ StreamerView.render("pleroma_respond.json", %{
	+ type: "pleroma.authenticate",
	+ result: "success"
	+ })}
	+ ], %{state \| user: user, oauth_token: oauth_token}}
	+ else
	+ {:auth, _, _} ->
	+ {[
	+ {:text,
	+ StreamerView.render("pleroma_respond.json", %{
	+ type: "pleroma.authenticate",
	+ result: "error",
	+ error: :already_authenticated
	+ })}
	+ ], state}
	+
	+ _ ->
	+ {[
	+ {:text,
	+ StreamerView.render("pleroma_respond.json", %{
	+ type: "pleroma.authenticate",
	+ result: "error",
	+ error: :unauthorized
	+ })}
	+ ], state}
	+ end
	+ end
	+
	defp handle_client_event(params, state) do
	Logger.error("#{__MODULE__} received unknown event: #{inspect(params)}")
	{[], state}
	end
	end
	diff --git a/lib/pleroma/web/views/streamer_view.ex b/lib/pleroma/web/views/streamer_view.ex
	index 19f098783..0cdcb1918 100644
	--- a/lib/pleroma/web/views/streamer_view.ex
	+++ b/lib/pleroma/web/views/streamer_view.ex
	@@ -1,156 +1,157 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.StreamerView do
	use Pleroma.Web, :view

	alias Pleroma.Activity
	alias Pleroma.Conversation.Participation
	alias Pleroma.Notification
	alias Pleroma.User
	alias Pleroma.Web.MastodonAPI.NotificationView

	def render("update.json", %Activity{} = activity, %User{} = user) do
	%{
	event: "update",
	payload:
	Pleroma.Web.MastodonAPI.StatusView.render(
	"show.json",
	activity: activity,
	for: user
	)
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("status_update.json", %Activity{} = activity, %User{} = user) do
	%{
	event: "status.update",
	payload:
	Pleroma.Web.MastodonAPI.StatusView.render(
	"show.json",
	activity: activity,
	for: user
	)
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("notification.json", %Notification{} = notify, %User{} = user) do
	%{
	event: "notification",
	payload:
	NotificationView.render(
	"show.json",
	%{notification: notify, for: user}
	)
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("update.json", %Activity{} = activity) do
	%{
	event: "update",
	payload:
	Pleroma.Web.MastodonAPI.StatusView.render(
	"show.json",
	activity: activity
	)
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("status_update.json", %Activity{} = activity) do
	%{
	event: "status.update",
	payload:
	Pleroma.Web.MastodonAPI.StatusView.render(
	"show.json",
	activity: activity
	)
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("chat_update.json", %{chat_message_reference: cm_ref}) do
	# Explicitly giving the cmr for the object here, so we don't accidentally
	# send a later 'last_message' that was inserted between inserting this and
	# streaming it out
	#
	# It also contains the chat with a cache of the correct unread count
	Logger.debug("Trying to stream out #{inspect(cm_ref)}")

	representation =
	Pleroma.Web.PleromaAPI.ChatView.render(
	"show.json",
	%{last_message: cm_ref, chat: cm_ref.chat}
	)

	%{
	event: "pleroma:chat_update",
	payload:
	representation
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("follow_relationships_update.json", item) do
	%{
	event: "pleroma:follow_relationships_update",
	payload:
	%{
	state: item.state,
	follower: %{
	id: item.follower.id,
	follower_count: item.follower.follower_count,
	following_count: item.follower.following_count
	},
	following: %{
	id: item.following.id,
	follower_count: item.following.follower_count,
	following_count: item.following.following_count
	}
	}
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("conversation.json", %Participation{} = participation) do
	%{
	event: "conversation",
	payload:
	Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{
	participation: participation,
	for: participation.user
	})
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("pleroma_respond.json", %{type: type, result: result} = params) do
	%{
	event: "pleroma.respond",
	payload:
	%{
	result: result,
	type: type
	}
	\|> Map.merge(maybe_error(params))
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	defp maybe_error(%{error: :bad_topic}), do: %{error: "bad_topic"}
	defp maybe_error(%{error: :unauthorized}), do: %{error: "unauthorized"}
	+ defp maybe_error(%{error: :already_authenticated}), do: %{error: "already_authenticated"}
	defp maybe_error(_), do: %{}
	end
	diff --git a/test/pleroma/integration/mastodon_websocket_test.exs b/test/pleroma/integration/mastodon_websocket_test.exs
	index 9db0f714f..827c7b5b0 100644
	--- a/test/pleroma/integration/mastodon_websocket_test.exs
	+++ b/test/pleroma/integration/mastodon_websocket_test.exs
	@@ -1,243 +1,324 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Integration.MastodonWebsocketTest do
	# Needs a streamer, needs to stay synchronous
	use Pleroma.DataCase

	import ExUnit.CaptureLog
	import Pleroma.Factory

	alias Pleroma.Integration.WebsocketClient
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.OAuth

	@moduletag needs_streamer: true, capture_log: true

	@path Pleroma.Web.Endpoint.url()
	\|> URI.parse()
	\|> Map.put(:scheme, "ws")
	\|> Map.put(:path, "/api/v1/streaming")
	\|> URI.to_string()

	def start_socket(qs \\ nil, headers \\ []) do
	path =
	case qs do
	nil -> @path
	qs -> @path <> qs
	end

	WebsocketClient.start_link(self(), path, headers)
	end

	defp decode_json(json) do
	with {:ok, %{"event" => event, "payload" => payload_text}} <- Jason.decode(json),
	{:ok, payload} <- Jason.decode(payload_text) do
	{:ok, %{"event" => event, "payload" => payload}}
	end
	end

	test "refuses invalid requests" do
	capture_log(fn ->
	assert {:error, %WebSockex.RequestError{code: 404}} = start_socket("?stream=ncjdk")
	Process.sleep(30)
	end)
	end

	test "requires authentication and a valid token for protected streams" do
	capture_log(fn ->
	assert {:error, %WebSockex.RequestError{code: 401}} =
	start_socket("?stream=user&access_token=aaaaaaaaaaaa")

	assert {:error, %WebSockex.RequestError{code: 401}} = start_socket("?stream=user")
	Process.sleep(30)
	end)
	end

	test "allows unified stream" do
	assert {:ok, _} = start_socket()
	end

	test "allows public streams without authentication" do
	assert {:ok, _} = start_socket("?stream=public")
	assert {:ok, _} = start_socket("?stream=public:local")
	assert {:ok, _} = start_socket("?stream=public:remote&instance=lain.com")
	assert {:ok, _} = start_socket("?stream=hashtag&tag=lain")
	end

	test "receives well formatted events" do
	user = insert(:user)
	{:ok, _} = start_socket("?stream=public")
	{:ok, activity} = CommonAPI.post(user, %{status: "nice echo chamber"})

	assert_receive {:text, raw_json}, 1_000
	assert {:ok, json} = Jason.decode(raw_json)

	assert "update" == json["event"]
	assert json["payload"]
	assert {:ok, json} = Jason.decode(json["payload"])

	view_json =
	Pleroma.Web.MastodonAPI.StatusView.render("show.json", activity: activity, for: nil)
	\|> Jason.encode!()
	\|> Jason.decode!()

	assert json == view_json
	end

	describe "subscribing via WebSocket" do
	test "can subscribe" do
	user = insert(:user)
	{:ok, pid} = start_socket()
	WebsocketClient.send_text(pid, %{type: "subscribe", stream: "public"} \|> Jason.encode!())
	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{"type" => "subscribe", "result" => "success"}
	}} = decode_json(raw_json)

	{:ok, activity} = CommonAPI.post(user, %{status: "nice echo chamber"})

	assert_receive {:text, raw_json}, 1_000
	assert {:ok, json} = Jason.decode(raw_json)

	assert "update" == json["event"]
	assert json["payload"]
	assert {:ok, json} = Jason.decode(json["payload"])

	view_json =
	Pleroma.Web.MastodonAPI.StatusView.render("show.json", activity: activity, for: nil)
	\|> Jason.encode!()
	\|> Jason.decode!()

	assert json == view_json
	end

	test "won't double subscribe" do
	user = insert(:user)
	{:ok, pid} = start_socket()
	WebsocketClient.send_text(pid, %{type: "subscribe", stream: "public"} \|> Jason.encode!())
	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{"type" => "subscribe", "result" => "success"}
	}} = decode_json(raw_json)

	WebsocketClient.send_text(pid, %{type: "subscribe", stream: "public"} \|> Jason.encode!())
	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{"type" => "subscribe", "result" => "ignored"}
	}} = decode_json(raw_json)

	{:ok, _activity} = CommonAPI.post(user, %{status: "nice echo chamber"})

	assert_receive {:text, _}, 1_000
	refute_receive {:text, _}, 1_000
	end

	test "can unsubscribe" do
	user = insert(:user)
	{:ok, pid} = start_socket()
	WebsocketClient.send_text(pid, %{type: "subscribe", stream: "public"} \|> Jason.encode!())
	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{"type" => "subscribe", "result" => "success"}
	}} = decode_json(raw_json)

	WebsocketClient.send_text(pid, %{type: "unsubscribe", stream: "public"} \|> Jason.encode!())
	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{"type" => "unsubscribe", "result" => "success"}
	}} = decode_json(raw_json)

	{:ok, _activity} = CommonAPI.post(user, %{status: "nice echo chamber"})
	refute_receive {:text, _}, 1_000
	end
	end

	describe "with a valid user token" do
	setup do
	{:ok, app} =
	Pleroma.Repo.insert(
	OAuth.App.register_changeset(%OAuth.App{}, %{
	client_name: "client",
	scopes: ["read"],
	redirect_uris: "url"
	})
	)

	user = insert(:user)

	{:ok, auth} = OAuth.Authorization.create_authorization(app, user)

	{:ok, token} = OAuth.Token.exchange_token(app, auth)

	%{app: app, user: user, token: token}
	end

	test "accepts valid tokens", state do
	assert {:ok, _} = start_socket("?stream=user&access_token=#{state.token.token}")
	end

	test "accepts the 'user' stream", %{token: token} = _state do
	assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")

	capture_log(fn ->
	assert {:error, %WebSockex.RequestError{code: 401}} = start_socket("?stream=user")
	Process.sleep(30)
	end)
	end

	test "accepts the 'user:notification' stream", %{token: token} = _state do
	assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")

	capture_log(fn ->
	assert {:error, %WebSockex.RequestError{code: 401}} =
	start_socket("?stream=user:notification")

	Process.sleep(30)
	end)
	end

	test "accepts valid token on Sec-WebSocket-Protocol header", %{token: token} do
	assert {:ok, _} = start_socket("?stream=user", [{"Sec-WebSocket-Protocol", token.token}])

	capture_log(fn ->
	assert {:error, %WebSockex.RequestError{code: 401}} =
	start_socket("?stream=user", [{"Sec-WebSocket-Protocol", "I am a friend"}])

	Process.sleep(30)
	end)
	end

	+ test "accepts valid token on client-sent event", %{token: token} do
	+ assert {:ok, pid} = start_socket()
	+
	+ WebsocketClient.send_text(
	+ pid,
	+ %{type: "pleroma.authenticate", token: token.token} \|> Jason.encode!()
	+ )
	+
	+ assert_receive {:text, raw_json}, 1_000
	+
	+ assert {:ok,
	+ %{
	+ "event" => "pleroma.respond",
	+ "payload" => %{"type" => "pleroma.authenticate", "result" => "success"}
	+ }} = decode_json(raw_json)
	+
	+ WebsocketClient.send_text(pid, %{type: "subscribe", stream: "user"} \|> Jason.encode!())
	+ assert_receive {:text, raw_json}, 1_000
	+
	+ assert {:ok,
	+ %{
	+ "event" => "pleroma.respond",
	+ "payload" => %{"type" => "subscribe", "result" => "success"}
	+ }} = decode_json(raw_json)
	+ end
	+
	+ test "rejects invalid token on client-sent event" do
	+ assert {:ok, pid} = start_socket()
	+
	+ WebsocketClient.send_text(
	+ pid,
	+ %{type: "pleroma.authenticate", token: "Something else"} \|> Jason.encode!()
	+ )
	+
	+ assert_receive {:text, raw_json}, 1_000
	+
	+ assert {:ok,
	+ %{
	+ "event" => "pleroma.respond",
	+ "payload" => %{
	+ "type" => "pleroma.authenticate",
	+ "result" => "error",
	+ "error" => "unauthorized"
	+ }
	+ }} = decode_json(raw_json)
	+ end
	+
	+ test "rejects new authenticate request if already logged-in", %{token: token} do
	+ assert {:ok, pid} = start_socket()
	+
	+ WebsocketClient.send_text(
	+ pid,
	+ %{type: "pleroma.authenticate", token: token.token} \|> Jason.encode!()
	+ )
	+
	+ assert_receive {:text, raw_json}, 1_000
	+
	+ assert {:ok,
	+ %{
	+ "event" => "pleroma.respond",
	+ "payload" => %{"type" => "pleroma.authenticate", "result" => "success"}
	+ }} = decode_json(raw_json)
	+
	+ WebsocketClient.send_text(
	+ pid,
	+ %{type: "pleroma.authenticate", token: "Something else"} \|> Jason.encode!()
	+ )
	+
	+ assert_receive {:text, raw_json}, 1_000
	+
	+ assert {:ok,
	+ %{
	+ "event" => "pleroma.respond",
	+ "payload" => %{
	+ "type" => "pleroma.authenticate",
	+ "result" => "error",
	+ "error" => "already_authenticated"
	+ }
	+ }} = decode_json(raw_json)
	+ end
	+
	test "disconnect when token is revoked", %{app: app, user: user, token: token} do
	assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")
	assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")

	{:ok, auth} = OAuth.Authorization.create_authorization(app, user)

	{:ok, token2} = OAuth.Token.exchange_token(app, auth)
	assert {:ok, _} = start_socket("?stream=user&access_token=#{token2.token}")

	OAuth.Token.Strategy.Revoke.revoke(token)

	assert_receive {:close, _}
	assert_receive {:close, _}
	refute_receive {:close, _}
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Thu, Oct 29, 11:03 PM (1 d, 14 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1798799
Default Alt Text: (23 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions