1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
5 defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
10 alias Pleroma.Web.OAuth.Token
11 alias Pleroma.Web.Streamer
13 @behaviour :cowboy_websocket
15 # Handled by periodic keepalive in Pleroma.Web.Streamer.Ping.
18 def init(%{qs: qs} = req, state) do
19 with params <- Enum.into(:cow_qs.parse_qs(qs), %{}),
20 sec_websocket <- :cowboy_req.header("sec-websocket-protocol", req, nil),
21 access_token <- Map.get(params, "access_token"),
22 {:ok, user} <- authenticate_request(access_token, sec_websocket),
23 {:ok, topic} <- Streamer.get_topic(Map.get(params, "stream"), user, params) do
26 :cowboy_req.set_resp_header("sec-websocket-protocol", sec_websocket, req)
31 {:cowboy_websocket, req, %{user: user, topic: topic}, %{idle_timeout: @timeout}}
33 {:error, :bad_topic} ->
34 Logger.debug("#{__MODULE__} bad topic #{inspect(req)}")
35 {:ok, req} = :cowboy_req.reply(404, req)
38 {:error, :unauthorized} ->
39 Logger.debug("#{__MODULE__} authentication error: #{inspect(req)}")
40 {:ok, req} = :cowboy_req.reply(401, req)
45 def websocket_init(state) do
46 send(self(), :subscribe)
50 # We never receive messages.
51 def websocket_handle(_frame, state) do
55 def websocket_info(:subscribe, state) do
57 "#{__MODULE__} accepted websocket connection for user #{
58 (state.user || %{id: "anonymous"}).id
59 }, topic #{state.topic}"
62 Streamer.add_socket(state.topic, streamer_socket(state))
66 def websocket_info({:text, message}, state) do
67 {:reply, {:text, message}, state}
70 def terminate(reason, _req, state) do
72 "#{__MODULE__} terminating websocket connection for user #{
73 (state.user || %{id: "anonymous"}).id
74 }, topic #{state.topic || "?"}: #{inspect(reason)}"
77 Streamer.remove_socket(state.topic, streamer_socket(state))
81 # Public streams without authentication.
82 defp authenticate_request(nil, nil) do
86 # Authenticated streams.
87 defp authenticate_request(access_token, sec_websocket) do
88 token = access_token || sec_websocket
90 with true <- is_bitstring(token),
91 %Token{user_id: user_id} <- Repo.get_by(Token, token: token),
92 user = %User{} <- User.get_cached_by_id(user_id) do
95 _ -> {:error, :unauthorized}
99 defp streamer_socket(state) do
100 %{transport_pid: self(), assigns: state}