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