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

   1 defmodule Pleroma.Web.Websub do
   2   alias Ecto.Changeset
   3   alias Pleroma.Repo
   4   alias Pleroma.Web.Websub.{WebsubServerSubscription, WebsubClientSubscription}
   5   alias Pleroma.Web.OStatus.FeedRepresenter
   6   alias Pleroma.Web.{XML, Endpoint, OStatus}
   7   alias Pleroma.Web.Router.Helpers
   8   require Logger
   9
  10   import Ecto.Query
  11
  12   @httpoison Application.get_env(:pleroma, :httpoison)
  13
  14   def verify(subscription, getter \\ &@httpoison.get/3) do
  15     challenge = Base.encode16(:crypto.strong_rand_bytes(8))
  16     lease_seconds = NaiveDateTime.diff(subscription.valid_until, subscription.updated_at)
  17     lease_seconds = lease_seconds |> to_string
  18
  19     params = %{
  20       "hub.challenge": challenge,
  21       "hub.lease_seconds": lease_seconds,
  22       "hub.topic": subscription.topic,
  23       "hub.mode": "subscribe"
  24     }
  25
  26     url = hd(String.split(subscription.callback, "?"))
  27     query = URI.parse(subscription.callback).query || ""
  28     params = Map.merge(params, URI.decode_query(query))
  29
  30     with {:ok, response} <- getter.(url, [], params: params),
  31          ^challenge <- response.body do
  32       changeset = Changeset.change(subscription, %{state: "active"})
  33       Repo.update(changeset)
  34     else
  35       e ->
  36         Logger.debug("Couldn't verify subscription")
  37         Logger.debug(inspect(e))
  38         {:error, subscription}
  39     end
  40   end
  41
  42   @supported_activities [
  43     "Create",
  44     "Follow",
  45     "Like",
  46     "Announce",
  47     "Undo",
  48     "Delete"
  49   ]
  50   def publish(topic, user, %{data: %{"type" => type}} = activity)
  51       when type in @supported_activities do
  52     # TODO: Only send to still valid subscriptions.
  53     query =
  54       from(
  55         sub in WebsubServerSubscription,
  56         where: sub.topic == ^topic and sub.state == "active",
  57         where: fragment("? > NOW()", sub.valid_until)
  58       )
  59
  60     subscriptions = Repo.all(query)
  61
  62     Enum.each(subscriptions, fn sub ->
  63       response =
  64         user
  65         |> FeedRepresenter.to_simple_form([activity], [user])
  66         |> :xmerl.export_simple(:xmerl_xml)
  67         |> to_string
  68
  69       data = %{
  70         xml: response,
  71         topic: topic,
  72         callback: sub.callback,
  73         secret: sub.secret
  74       }
  75
  76       Pleroma.Web.Federator.enqueue(:publish_single_websub, data)
  77     end)
  78   end
  79
  80   def publish(_, _, _), do: ""
  81
  82   def sign(secret, doc) do
  83     :crypto.hmac(:sha, secret, to_string(doc)) |> Base.encode16() |> String.downcase()
  84   end
  85
  86   def incoming_subscription_request(user, %{"hub.mode" => "subscribe"} = params) do
  87     with {:ok, topic} <- valid_topic(params, user),
  88          {:ok, lease_time} <- lease_time(params),
  89          secret <- params["hub.secret"],
  90          callback <- params["hub.callback"] do
  91       subscription = get_subscription(topic, callback)
  92
  93       data = %{
  94         state: subscription.state || "requested",
  95         topic: topic,
  96         secret: secret,
  97         callback: callback
  98       }
  99
 100       change = Changeset.change(subscription, data)
 101       websub = Repo.insert_or_update!(change)
 102
 103       change =
 104         Changeset.change(websub, %{valid_until: NaiveDateTime.add(websub.updated_at, lease_time)})
 105
 106       websub = Repo.update!(change)
 107
 108       Pleroma.Web.Federator.enqueue(:verify_websub, websub)
 109
 110       {:ok, websub}
 111     else
 112       {:error, reason} ->
 113         Logger.debug("Couldn't create subscription")
 114         Logger.debug(inspect(reason))
 115
 116         {:error, reason}
 117     end
 118   end
 119
 120   defp get_subscription(topic, callback) do
 121     Repo.get_by(WebsubServerSubscription, topic: topic, callback: callback) ||
 122       %WebsubServerSubscription{}
 123   end
 124
 125   # Temp hack for mastodon.
 126   defp lease_time(%{"hub.lease_seconds" => ""}) do
 127     # three days
 128     {:ok, 60 * 60 * 24 * 3}
 129   end
 130
 131   defp lease_time(%{"hub.lease_seconds" => lease_seconds}) do
 132     {:ok, String.to_integer(lease_seconds)}
 133   end
 134
 135   defp lease_time(_) do
 136     # three days
 137     {:ok, 60 * 60 * 24 * 3}
 138   end
 139
 140   defp valid_topic(%{"hub.topic" => topic}, user) do
 141     if topic == OStatus.feed_path(user) do
 142       {:ok, OStatus.feed_path(user)}
 143     else
 144       {:error, "Wrong topic requested, expected #{OStatus.feed_path(user)}, got #{topic}"}
 145     end
 146   end
 147
 148   def subscribe(subscriber, subscribed, requester \\ &request_subscription/1) do
 149     topic = subscribed.info["topic"]
 150     # FIXME: Race condition, use transactions
 151     {:ok, subscription} =
 152       with subscription when not is_nil(subscription) <-
 153              Repo.get_by(WebsubClientSubscription, topic: topic) do
 154         subscribers = [subscriber.ap_id | subscription.subscribers] |> Enum.uniq()
 155         change = Ecto.Changeset.change(subscription, %{subscribers: subscribers})
 156         Repo.update(change)
 157       else
 158         _e ->
 159           subscription = %WebsubClientSubscription{
 160             topic: topic,
 161             hub: subscribed.info["hub"],
 162             subscribers: [subscriber.ap_id],
 163             state: "requested",
 164             secret: :crypto.strong_rand_bytes(8) |> Base.url_encode64(),
 165             user: subscribed
 166           }
 167
 168           Repo.insert(subscription)
 169       end
 170
 171     requester.(subscription)
 172   end
 173
 174   def gather_feed_data(topic, getter \\ &@httpoison.get/1) do
 175     with {:ok, response} <- getter.(topic),
 176          status_code when status_code in 200..299 <- response.status_code,
 177          body <- response.body,
 178          doc <- XML.parse_document(body),
 179          uri when not is_nil(uri) <- XML.string_from_xpath("/feed/author[1]/uri", doc),
 180          hub when not is_nil(hub) <- XML.string_from_xpath(~S{/feed/link[@rel="hub"]/@href}, doc) do
 181       name = XML.string_from_xpath("/feed/author[1]/name", doc)
 182       preferredUsername = XML.string_from_xpath("/feed/author[1]/poco:preferredUsername", doc)
 183       displayName = XML.string_from_xpath("/feed/author[1]/poco:displayName", doc)
 184       avatar = OStatus.make_avatar_object(doc)
 185       bio = XML.string_from_xpath("/feed/author[1]/summary", doc)
 186
 187       {:ok,
 188        %{
 189          "uri" => uri,
 190          "hub" => hub,
 191          "nickname" => preferredUsername || name,
 192          "name" => displayName || name,
 193          "host" => URI.parse(uri).host,
 194          "avatar" => avatar,
 195          "bio" => bio
 196        }}
 197     else
 198       e ->
 199         {:error, e}
 200     end
 201   end
 202
 203   def request_subscription(websub, poster \\ &@httpoison.post/3, timeout \\ 10_000) do
 204     data = [
 205       "hub.mode": "subscribe",
 206       "hub.topic": websub.topic,
 207       "hub.secret": websub.secret,
 208       "hub.callback": Helpers.websub_url(Endpoint, :websub_subscription_confirmation, websub.id)
 209     ]
 210
 211     # This checks once a second if we are confirmed yet
 212     websub_checker = fn ->
 213       helper = fn helper ->
 214         :timer.sleep(1000)
 215         websub = Repo.get_by(WebsubClientSubscription, id: websub.id, state: "accepted")
 216         if websub, do: websub, else: helper.(helper)
 217       end
 218
 219       helper.(helper)
 220     end
 221
 222     task = Task.async(websub_checker)
 223
 224     with {:ok, %{status_code: 202}} <-
 225            poster.(websub.hub, {:form, data}, "Content-type": "application/x-www-form-urlencoded"),
 226          {:ok, websub} <- Task.yield(task, timeout) do
 227       {:ok, websub}
 228     else
 229       e ->
 230         Task.shutdown(task)
 231
 232         change = Ecto.Changeset.change(websub, %{state: "rejected"})
 233         {:ok, websub} = Repo.update(change)
 234
 235         Logger.debug(fn -> "Couldn't confirm subscription: #{inspect(websub)}" end)
 236         Logger.debug(fn -> "error: #{inspect(e)}" end)
 237
 238         {:error, websub}
 239     end
 240   end
 241
 242   def refresh_subscriptions(delta \\ 60 * 60 * 24) do
 243     Logger.debug("Refreshing subscriptions")
 244
 245     cut_off = NaiveDateTime.add(NaiveDateTime.utc_now(), delta)
 246
 247     query = from(sub in WebsubClientSubscription, where: sub.valid_until < ^cut_off)
 248
 249     subs = Repo.all(query)
 250
 251     Enum.each(subs, fn sub ->
 252       Pleroma.Web.Federator.enqueue(:request_subscription, sub)
 253     end)
 254   end
 255
 256   def publish_one(%{xml: xml, topic: topic, callback: callback, secret: secret}) do
 257     signature = sign(secret || "", xml)
 258     Logger.info(fn -> "Pushing #{topic} to #{callback}" end)
 259
 260     with {:ok, %{status_code: code}} <-
 261            @httpoison.post(
 262              callback,
 263              xml,
 264              [
 265                {"Content-Type", "application/atom+xml"},
 266                {"X-Hub-Signature", "sha1=#{signature}"}
 267              ],
 268              timeout: 10000,
 269              recv_timeout: 20000,
 270              hackney: [pool: :default]
 271            ) do
 272       Logger.info(fn -> "Pushed to #{callback}, code #{code}" end)
 273       {:ok, code}
 274     else
 275       e ->
 276         Logger.debug(fn -> "Couldn't push to #{callback}, #{inspect(e)}" end)
 277         {:error, e}
 278     end
 279   end
 280 end