git.squeep.com Git - akkoma/blob - lib/pleroma/web/fed_sockets/outgoing_handler.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Web.FedSockets.OutgoingHandler do
   6   use GenServer
   7
   8   require Logger
   9
  10   alias Pleroma.Web.ActivityPub.InternalFetchActor
  11   alias Pleroma.Web.FedSockets
  12   alias Pleroma.Web.FedSockets.FedRegistry
  13   alias Pleroma.Web.FedSockets.FedSocket
  14   alias Pleroma.Web.FedSockets.SocketInfo
  15
  16   def start_link(uri) do
  17     GenServer.start_link(__MODULE__, %{uri: uri})
  18   end
  19
  20   def init(%{uri: uri}) do
  21     case initiate_connection(uri) do
  22       {:ok, ws_origin, conn_pid} ->
  23         FedRegistry.add_fed_socket(ws_origin, conn_pid)
  24
  25       {:error, reason} ->
  26         Logger.debug("Outgoing connection failed - #{inspect(reason)}")
  27         :ignore
  28     end
  29   end
  30
  31   def handle_info({:gun_ws, conn_pid, _ref, {:text, data}}, socket_info) do
  32     socket_info = SocketInfo.touch(socket_info)
  33
  34     case FedSocket.receive_package(socket_info, data) do
  35       {:noreply, _} ->
  36         {:noreply, socket_info}
  37
  38       {:reply, reply} ->
  39         :gun.ws_send(conn_pid, {:text, Jason.encode!(reply)})
  40         {:noreply, socket_info}
  41
  42       {:error, reason} ->
  43         Logger.error("incoming error - receive_package: #{inspect(reason)}")
  44         {:noreply, socket_info}
  45     end
  46   end
  47
  48   def handle_info(:close, state) do
  49     Logger.debug("Sending close frame !!!!!!!")
  50     {:close, state}
  51   end
  52
  53   def handle_info({:gun_down, _pid, _prot, :closed, _}, state) do
  54     {:stop, :normal, state}
  55   end
  56
  57   def handle_info({:send, data}, %{conn_pid: conn_pid} = socket_info) do
  58     socket_info = SocketInfo.touch(socket_info)
  59     :gun.ws_send(conn_pid, {:text, data})
  60     {:noreply, socket_info}
  61   end
  62
  63   def handle_info({:gun_ws, _, _, :pong}, state) do
  64     {:noreply, state, :hibernate}
  65   end
  66
  67   def handle_info(msg, state) do
  68     Logger.debug("#{__MODULE__} unhandled event #{inspect(msg)}")
  69     {:noreply, state}
  70   end
  71
  72   def terminate(reason, state) do
  73     Logger.debug(
  74       "#{__MODULE__} terminating outgoing connection for #{inspect(state)} for #{inspect(reason)}"
  75     )
  76
  77     {:ok, state}
  78   end
  79
  80   def initiate_connection(uri) do
  81     ws_uri =
  82       uri
  83       |> SocketInfo.origin()
  84       |> FedSockets.uri_for_origin()
  85
  86     %{host: host, port: port, path: path} = URI.parse(ws_uri)
  87
  88     with {:ok, conn_pid} <- :gun.open(to_charlist(host), port),
  89          {:ok, _} <- :gun.await_up(conn_pid),
  90          reference <- :gun.get(conn_pid, to_charlist(path)),
  91          {:response, :fin, 204, _} <- :gun.await(conn_pid, reference),
  92          headers <- build_headers(uri),
  93          ref <- :gun.ws_upgrade(conn_pid, to_charlist(path), headers, %{silence_pings: false}) do
  94       receive do
  95         {:gun_upgrade, ^conn_pid, ^ref, [<<"websocket">>], _} ->
  96           {:ok, ws_uri, conn_pid}
  97       after
  98         15_000 ->
  99           Logger.debug("Fedsocket timeout connecting to #{inspect(uri)}")
 100           {:error, :timeout}
 101       end
 102     else
 103       {:response, :nofin, 404, _} ->
 104         {:error, :fedsockets_not_supported}
 105
 106       e ->
 107         Logger.debug("Fedsocket error connecting to #{inspect(uri)}")
 108         {:error, e}
 109     end
 110   end
 111
 112   defp build_headers(uri) do
 113     host_for_sig = uri |> URI.parse() |> host_signature()
 114
 115     shake = FedSocket.shake()
 116     digest = "SHA-256=" <> (:crypto.hash(:sha256, shake) |> Base.encode64())
 117     date = Pleroma.Signature.signed_date()
 118     shake_size = byte_size(shake)
 119
 120     signature_opts = %{
 121       "(request-target)": shake,
 122       "content-length": to_charlist("#{shake_size}"),
 123       date: date,
 124       digest: digest,
 125       host: host_for_sig
 126     }
 127
 128     signature = Pleroma.Signature.sign(InternalFetchActor.get_actor(), signature_opts)
 129
 130     [
 131       {'signature', to_charlist(signature)},
 132       {'date', date},
 133       {'digest', to_charlist(digest)},
 134       {'content-length', to_charlist("#{shake_size}")},
 135       {to_charlist("(request-target)"), to_charlist(shake)}
 136     ]
 137   end
 138
 139   defp host_signature(%{host: host, scheme: scheme, port: port}) do
 140     if port == URI.default_port(scheme) do
 141       host
 142     else
 143       "#{host}:#{port}"
 144     end
 145   end
 146 end