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