git.squeep.com Git - akkoma/blob - lib/pleroma/web/mastodon_api/websocket_handler.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
   6   require Logger
   7
   8   alias Pleroma.Repo
   9   alias Pleroma.User
  10   alias Pleroma.Web.OAuth.Token
  11   alias Pleroma.Web.Streamer
  12
  13   @behaviour :cowboy_websocket
  14
  15   @streams [
  16     "public",
  17     "public:local",
  18     "public:media",
  19     "public:local:media",
  20     "user",
  21     "user:notification",
  22     "direct",
  23     "list",
  24     "hashtag"
  25   ]
  26   @anonymous_streams ["public", "public:local", "hashtag"]
  27
  28   # Handled by periodic keepalive in Pleroma.Web.Streamer.Ping.
  29   @timeout :infinity
  30
  31   def init(%{qs: qs} = req, state) do
  32     with params <- :cow_qs.parse_qs(qs),
  33          sec_websocket <- :cowboy_req.header("sec-websocket-protocol", req, nil),
  34          access_token <- List.keyfind(params, "access_token", 0),
  35          {_, stream} <- List.keyfind(params, "stream", 0),
  36          {:ok, user} <- allow_request(stream, [access_token, sec_websocket]),
  37          topic when is_binary(topic) <- expand_topic(stream, params) do
  38       {:cowboy_websocket, req, %{user: user, topic: topic}, %{idle_timeout: @timeout}}
  39     else
  40       {:error, code} ->
  41         Logger.debug("#{__MODULE__} denied connection: #{inspect(code)} - #{inspect(req)}")
  42         {:ok, req} = :cowboy_req.reply(code, req)
  43         {:ok, req, state}
  44
  45       error ->
  46         Logger.debug("#{__MODULE__} denied connection: #{inspect(error)} - #{inspect(req)}")
  47         {:ok, req} = :cowboy_req.reply(400, req)
  48         {:ok, req, state}
  49     end
  50   end
  51
  52   def websocket_init(state) do
  53     send(self(), :subscribe)
  54     {:ok, state}
  55   end
  56
  57   # We never receive messages.
  58   def websocket_handle(_frame, state) do
  59     {:ok, state}
  60   end
  61
  62   def websocket_info(:subscribe, state) do
  63     Logger.debug(
  64       "#{__MODULE__} accepted websocket connection for user #{
  65         (state.user || %{id: "anonymous"}).id
  66       }, topic #{state.topic}"
  67     )
  68
  69     Streamer.add_socket(state.topic, streamer_socket(state))
  70     {:ok, state}
  71   end
  72
  73   def websocket_info({:text, message}, state) do
  74     {:reply, {:text, message}, state}
  75   end
  76
  77   def terminate(reason, _req, state) do
  78     Logger.debug(
  79       "#{__MODULE__} terminating websocket connection for user #{
  80         (state.user || %{id: "anonymous"}).id
  81       }, topic #{state.topic || "?"}: #{inspect(reason)}"
  82     )
  83
  84     Streamer.remove_socket(state.topic, streamer_socket(state))
  85     :ok
  86   end
  87
  88   # Public streams without authentication.
  89   defp allow_request(stream, [nil, nil]) when stream in @anonymous_streams do
  90     {:ok, nil}
  91   end
  92
  93   # Authenticated streams.
  94   defp allow_request(stream, [access_token, sec_websocket]) when stream in @streams do
  95     token =
  96       with {"access_token", token} <- access_token do
  97         token
  98       else
  99         _ -> sec_websocket
 100       end
 101
 102     with true <- is_bitstring(token),
 103          %Token{user_id: user_id} <- Repo.get_by(Token, token: token),
 104          user = %User{} <- User.get_cached_by_id(user_id) do
 105       {:ok, user}
 106     else
 107       _ -> {:error, 403}
 108     end
 109   end
 110
 111   # Not authenticated.
 112   defp allow_request(stream, _) when stream in @streams, do: {:error, 403}
 113
 114   # No matching stream.
 115   defp allow_request(_, _), do: {:error, 404}
 116
 117   defp expand_topic("hashtag", params) do
 118     case List.keyfind(params, "tag", 0) do
 119       {_, tag} -> "hashtag:#{tag}"
 120       _ -> nil
 121     end
 122   end
 123
 124   defp expand_topic("list", params) do
 125     case List.keyfind(params, "list", 0) do
 126       {_, list} -> "list:#{list}"
 127       _ -> nil
 128     end
 129   end
 130
 131   defp expand_topic(topic, _), do: topic
 132
 133   defp streamer_socket(state) do
 134     %{transport_pid: self(), assigns: state}
 135   end
 136 end