git.squeep.com Git - akkoma/blob - lib/pleroma/web/websub/websub.ex

   1 defmodule Pleroma.Web.Websub do
   2   alias Pleroma.Repo
   3   alias Pleroma.Web.Websub.{WebsubServerSubscription, WebsubClientSubscription}
   4   alias Pleroma.Web.OStatus.FeedRepresenter
   5   alias Pleroma.Web.OStatus
   6   alias Pleroma.Web.XML
   7
   8   import Ecto.Query
   9
  10   def verify(subscription, getter \\ &HTTPoison.get/3 ) do
  11     challenge = Base.encode16(:crypto.strong_rand_bytes(8))
  12     lease_seconds = NaiveDateTime.diff(subscription.valid_until, subscription.updated_at) |> to_string
  13
  14     params = %{
  15       "hub.challenge": challenge,
  16       "hub.lease_seconds": lease_seconds,
  17       "hub.topic": subscription.topic,
  18       "hub.mode": "subscribe"
  19     }
  20
  21     url = hd(String.split(subscription.callback, "?"))
  22     query = URI.parse(subscription.callback).query || ""
  23     params = Map.merge(params, URI.decode_query(query))
  24     with {:ok, response} <- getter.(url, [], [params: params]),
  25          ^challenge <- response.body
  26     do
  27       changeset = Ecto.Changeset.change(subscription, %{state: "active"})
  28       Repo.update(changeset)
  29     else _e ->
  30       changeset = Ecto.Changeset.change(subscription, %{state: "rejected"})
  31       {:ok, subscription } = Repo.update(changeset)
  32       {:error, subscription}
  33     end
  34   end
  35
  36   def publish(topic, user, activity) do
  37     query = from sub in WebsubServerSubscription,
  38     where: sub.topic == ^topic and sub.state == "active"
  39     subscriptions = Repo.all(query)
  40     Enum.each(subscriptions, fn(sub) ->
  41       response = FeedRepresenter.to_simple_form(user, [activity], [user])
  42       |> :xmerl.export_simple(:xmerl_xml)
  43
  44       signature = :crypto.hmac(:sha, sub.secret, response) |> Base.encode16
  45
  46       HTTPoison.post(sub.callback, response, [
  47             {"Content-Type", "application/atom+xml"},
  48             {"X-Hub-Signature", "sha1=#{signature}"}
  49           ])
  50     end)
  51   end
  52
  53   def incoming_subscription_request(user, %{"hub.mode" => "subscribe"} = params) do
  54     with {:ok, topic} <- valid_topic(params, user),
  55          {:ok, lease_time} <- lease_time(params),
  56          secret <- params["hub.secret"],
  57          callback <- params["hub.callback"]
  58     do
  59       subscription = get_subscription(topic, callback)
  60       data = %{
  61         state: subscription.state || "requested",
  62         topic: topic,
  63         secret: secret,
  64         callback: callback
  65       }
  66
  67       change = Ecto.Changeset.change(subscription, data)
  68       websub = Repo.insert_or_update!(change)
  69
  70       change = Ecto.Changeset.change(websub, %{valid_until: NaiveDateTime.add(websub.updated_at, lease_time)})
  71       websub = Repo.update!(change)
  72
  73       Pleroma.Web.Federator.enqueue(:verify_websub, websub)
  74
  75       {:ok, websub}
  76     else {:error, reason} ->
  77       {:error, reason}
  78     end
  79   end
  80
  81   defp get_subscription(topic, callback) do
  82     Repo.get_by(WebsubServerSubscription, topic: topic, callback: callback) || %WebsubServerSubscription{}
  83   end
  84
  85   defp lease_time(%{"hub.lease_seconds" => lease_seconds}) do
  86     {:ok, String.to_integer(lease_seconds)}
  87   end
  88
  89   defp lease_time(_) do
  90     {:ok, 60 * 60 * 24 * 3} # three days
  91   end
  92
  93   defp valid_topic(%{"hub.topic" => topic}, user) do
  94     if topic == OStatus.feed_path(user) do
  95       {:ok, topic}
  96     else
  97       {:error, "Wrong topic requested, expected #{OStatus.feed_path(user)}, got #{topic}"}
  98     end
  99   end
 100
 101   def subscribe(user, topic) do
 102     # Race condition, use transactions
 103     {:ok, subscription} = with subscription when not is_nil(subscription) <- Repo.get_by(WebsubClientSubscription, topic: topic) do
 104       subscribers = [user.ap_id, subscription.subcribers] |> Enum.uniq
 105       change = Ecto.Changeset.change(subscription, %{subscribers: subscribers})
 106       Repo.update(change)
 107     else _e ->
 108       subscription = %WebsubClientSubscription{
 109         topic: topic,
 110         subscribers: [user.ap_id],
 111         state: "requested",
 112         secret: :crypto.strong_rand_bytes(8) |> Base.url_encode64
 113       }
 114       Repo.insert(subscription)
 115     end
 116
 117     {:ok, subscription}
 118   end
 119 end