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