1 defmodule Pleroma.Web.Websub do
3 alias Pleroma.Web.Websub.{WebsubServerSubscription, WebsubClientSubscription}
4 alias Pleroma.Web.OStatus.FeedRepresenter
5 alias Pleroma.Web.{XML, Endpoint, OStatus}
6 alias Pleroma.Web.Router.Helpers
11 def verify(subscription, getter \\ &HTTPoison.get/3 ) do
12 challenge = Base.encode16(:crypto.strong_rand_bytes(8))
13 lease_seconds = NaiveDateTime.diff(subscription.valid_until, subscription.updated_at) |> to_string
16 "hub.challenge": challenge,
17 "hub.lease_seconds": lease_seconds,
18 "hub.topic": subscription.topic,
19 "hub.mode": "subscribe"
22 url = hd(String.split(subscription.callback, "?"))
23 query = URI.parse(subscription.callback).query || ""
24 params = Map.merge(params, URI.decode_query(query))
25 with {:ok, response} <- getter.(url, [], [params: params]),
26 ^challenge <- response.body
28 changeset = Ecto.Changeset.change(subscription, %{state: "active"})
29 Repo.update(changeset)
31 changeset = Ecto.Changeset.change(subscription, %{state: "rejected"})
32 {:ok, subscription } = Repo.update(changeset)
33 {:error, subscription}
37 def publish(topic, user, activity) do
38 query = from sub in WebsubServerSubscription,
39 where: sub.topic == ^topic and sub.state == "active"
40 subscriptions = Repo.all(query)
41 Enum.each(subscriptions, fn(sub) ->
42 response = FeedRepresenter.to_simple_form(user, [activity], [user])
43 |> :xmerl.export_simple(:xmerl_xml)
45 signature = sign(sub.secret, response)
46 HTTPoison.post(sub.callback, response, [
47 {"Content-Type", "application/atom+xml"},
48 {"X-Hub-Signature", "sha1=#{signature}"}
53 def sign(secret, doc) do
54 :crypto.hmac(:sha, secret, doc) |> Base.encode16
57 def incoming_subscription_request(user, %{"hub.mode" => "subscribe"} = params) do
58 with {:ok, topic} <- valid_topic(params, user),
59 {:ok, lease_time} <- lease_time(params),
60 secret <- params["hub.secret"],
61 callback <- params["hub.callback"]
63 subscription = get_subscription(topic, callback)
65 state: subscription.state || "requested",
71 change = Ecto.Changeset.change(subscription, data)
72 websub = Repo.insert_or_update!(change)
74 change = Ecto.Changeset.change(websub, %{valid_until: NaiveDateTime.add(websub.updated_at, lease_time)})
75 websub = Repo.update!(change)
77 Pleroma.Web.Federator.enqueue(:verify_websub, websub)
80 else {:error, reason} ->
81 Logger.debug("Couldn't create subscription.")
82 Logger.debug(inspect(reason))
88 defp get_subscription(topic, callback) do
89 Repo.get_by(WebsubServerSubscription, topic: topic, callback: callback) || %WebsubServerSubscription{}
92 defp lease_time(%{"hub.lease_seconds" => lease_seconds}) do
93 {:ok, String.to_integer(lease_seconds)}
97 {:ok, 60 * 60 * 24 * 3} # three days
100 defp valid_topic(%{"hub.topic" => topic}, user) do
101 if topic == OStatus.feed_path(user) do
104 {:error, "Wrong topic requested, expected #{OStatus.feed_path(user)}, got #{topic}"}
108 def subscribe(subscriber, subscribed, requester \\ &request_subscription/1) do
109 topic = subscribed.info["topic"]
110 # FIXME: Race condition, use transactions
111 {:ok, subscription} = with subscription when not is_nil(subscription) <- Repo.get_by(WebsubClientSubscription, topic: topic) do
112 subscribers = [subscriber.ap_id, subscription.subscribers] |> Enum.uniq
113 change = Ecto.Changeset.change(subscription, %{subscribers: subscribers})
116 subscription = %WebsubClientSubscription{
118 hub: subscribed.info["hub"],
119 subscribers: [subscriber.ap_id],
121 secret: :crypto.strong_rand_bytes(8) |> Base.url_encode64,
124 Repo.insert(subscription)
126 requester.(subscription)
129 def gather_feed_data(topic, getter \\ &HTTPoison.get/1) do
130 with {:ok, response} <- getter.(topic),
131 status_code when status_code in 200..299 <- response.status_code,
132 body <- response.body,
133 doc <- XML.parse_document(body),
134 uri when not is_nil(uri) <- XML.string_from_xpath("/feed/author[1]/uri", doc),
135 hub when not is_nil(hub) <- XML.string_from_xpath(~S{/feed/link[@rel="hub"]/@href}, doc) do
137 name = XML.string_from_xpath("/feed/author[1]/name", doc)
138 preferredUsername = XML.string_from_xpath("/feed/author[1]/poco:preferredUsername", doc)
139 displayName = XML.string_from_xpath("/feed/author[1]/poco:displayName", doc)
140 avatar = OStatus.make_avatar_object(doc)
145 nickname: preferredUsername || name,
146 name: displayName || name,
147 host: URI.parse(uri).host,
155 def request_subscription(websub, poster \\ &HTTPoison.post/3, timeout \\ 10_000) do
157 "hub.mode": "subscribe",
158 "hub.topic": websub.topic,
159 "hub.secret": websub.secret,
160 "hub.callback": Helpers.websub_url(Endpoint, :websub_subscription_confirmation, websub.id)
163 # This checks once a second if we are confirmed yet
164 websub_checker = fn ->
165 helper = fn (helper) ->
167 websub = Repo.get_by(WebsubClientSubscription, id: websub.id, state: "accepted")
168 if websub, do: websub, else: helper.(helper)
173 task = Task.async(websub_checker)
175 with {:ok, %{status_code: 202}} <- poster.(websub.hub, {:form, data}, ["Content-type": "application/x-www-form-urlencoded"]),
176 {:ok, websub} <- Task.yield(task, timeout) do
181 change = Ecto.Changeset.change(websub, %{state: "rejected"})
182 {:ok, websub} = Repo.update(change)
184 Logger.debug("Couldn't confirm subscription: #{inspect(websub)}")
185 Logger.debug("error: #{inspect(e)}")