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   defp get_subscription(topic, callback) do
 125     Repo.get_by(WebsubServerSubscription, topic: topic, callback: callback) ||
 126       %WebsubServerSubscription{}
 127   end
 128
 129   # Temp hack for mastodon.
 130   defp lease_time(%{"hub.lease_seconds" => ""}) do
 131     # three days
 132     {:ok, 60 * 60 * 24 * 3}
 133   end
 134
 135   defp lease_time(%{"hub.lease_seconds" => lease_seconds}) do
 136     {:ok, String.to_integer(lease_seconds)}
 137   end
 138
 139   defp lease_time(_) do
 140     # three days
 141     {:ok, 60 * 60 * 24 * 3}
 142   end
 143
 144   defp valid_topic(%{"hub.topic" => topic}, user) do
 145     if topic == OStatus.feed_path(user) do
 146       {:ok, OStatus.feed_path(user)}
 147     else
 148       {:error, "Wrong topic requested, expected #{OStatus.feed_path(user)}, got #{topic}"}
 149     end
 150   end
 151
 152   def subscribe(subscriber, subscribed, requester \\ &request_subscription/1) do
 153     topic = subscribed.info.topic
 154     # FIXME: Race condition, use transactions
 155     {:ok, subscription} =
 156       with subscription when not is_nil(subscription) <-
 157              Repo.get_by(WebsubClientSubscription, topic: topic) do
 158         subscribers = [subscriber.ap_id | subscription.subscribers] |> Enum.uniq()
 159         change = Ecto.Changeset.change(subscription, %{subscribers: subscribers})
 160         Repo.update(change)
 161       else
 162         _e ->
 163           subscription = %WebsubClientSubscription{
 164             topic: topic,
 165             hub: subscribed.info.hub,
 166             subscribers: [subscriber.ap_id],
 167             state: "requested",
 168             secret: :crypto.strong_rand_bytes(8) |> Base.url_encode64(),
 169             user: subscribed
 170           }
 171
 172           Repo.insert(subscription)
 173       end
 174
 175     requester.(subscription)
 176   end
 177
 178   def gather_feed_data(topic, getter \\ &@httpoison.get/1) do
 179     with {:ok, response} <- getter.(topic),
 180          status when status in 200..299 <- response.status,
 181          body <- response.body,
 182          doc <- XML.parse_document(body),
 183          uri when not is_nil(uri) <- XML.string_from_xpath("/feed/author[1]/uri", doc),
 184          hub when not is_nil(hub) <- XML.string_from_xpath(~S{/feed/link[@rel="hub"]/@href}, doc) do
 185       name = XML.string_from_xpath("/feed/author[1]/name", doc)
 186       preferredUsername = XML.string_from_xpath("/feed/author[1]/poco:preferredUsername", doc)
 187       displayName = XML.string_from_xpath("/feed/author[1]/poco:displayName", doc)
 188       avatar = OStatus.make_avatar_object(doc)
 189       bio = XML.string_from_xpath("/feed/author[1]/summary", doc)
 190
 191       {:ok,
 192        %{
 193          "uri" => uri,
 194          "hub" => hub,
 195          "nickname" => preferredUsername || name,
 196          "name" => displayName || name,
 197          "host" => URI.parse(uri).host,
 198          "avatar" => avatar,
 199          "bio" => bio
 200        }}
 201     else
 202       e ->
 203         {:error, e}
 204     end
 205   end
 206
 207   def request_subscription(websub, poster \\ &@httpoison.post/3, timeout \\ 10_000) do
 208     data = [
 209       "hub.mode": "subscribe",
 210       "hub.topic": websub.topic,
 211       "hub.secret": websub.secret,
 212       "hub.callback": Helpers.websub_url(Endpoint, :websub_subscription_confirmation, websub.id)
 213     ]
 214
 215     # This checks once a second if we are confirmed yet
 216     websub_checker = fn ->
 217       helper = fn helper ->
 218         :timer.sleep(1000)
 219         websub = Repo.get_by(WebsubClientSubscription, id: websub.id, state: "accepted")
 220         if websub, do: websub, else: helper.(helper)
 221       end
 222
 223       helper.(helper)
 224     end
 225
 226     task = Task.async(websub_checker)
 227
 228     with {:ok, %{status: 202}} <-
 229            poster.(websub.hub, {:form, data}, "Content-type": "application/x-www-form-urlencoded"),
 230          {:ok, websub} <- Task.yield(task, timeout) do
 231       {:ok, websub}
 232     else
 233       e ->
 234         Task.shutdown(task)
 235
 236         change = Ecto.Changeset.change(websub, %{state: "rejected"})
 237         {:ok, websub} = Repo.update(change)
 238
 239         Logger.debug(fn -> "Couldn't confirm subscription: #{inspect(websub)}" end)
 240         Logger.debug(fn -> "error: #{inspect(e)}" end)
 241
 242         {:error, websub}
 243     end
 244   end
 245
 246   def refresh_subscriptions(delta \\ 60 * 60 * 24) do
 247     Logger.debug("Refreshing subscriptions")
 248
 249     cut_off = NaiveDateTime.add(NaiveDateTime.utc_now(), delta)
 250
 251     query = from(sub in WebsubClientSubscription, where: sub.valid_until < ^cut_off)
 252
 253     subs = Repo.all(query)
 254
 255     Enum.each(subs, fn sub ->
 256       Pleroma.Web.Federator.enqueue(:request_subscription, sub)
 257     end)
 258   end
 259
 260   def publish_one(%{xml: xml, topic: topic, callback: callback, secret: secret}) do
 261     signature = sign(secret || "", xml)
 262     Logger.info(fn -> "Pushing #{topic} to #{callback}" end)
 263
 264     with {:ok, %{status: code}} <-
 265            @httpoison.post(
 266              callback,
 267              xml,
 268              [
 269                {"Content-Type", "application/atom+xml"},
 270                {"X-Hub-Signature", "sha1=#{signature}"}
 271              ]
 272            ) do
 273       Logger.info(fn -> "Pushed to #{callback}, code #{code}" end)
 274       {:ok, code}
 275     else
 276       e ->
 277         Logger.debug(fn -> "Couldn't push to #{callback}, #{inspect(e)}" end)
 278         {:error, e}
 279     end
 280   end
 281 end