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