No OneTemporary
Actions

Authored By

Unknown

Size

12 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/web/websub/websub.ex b/lib/pleroma/web/websub/websub.ex
	index 3a287edd9..7ca62c83b 100644
	--- a/lib/pleroma/web/websub/websub.ex
	+++ b/lib/pleroma/web/websub/websub.ex
	@@ -1,281 +1,287 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Websub do
	alias Ecto.Changeset
	alias Pleroma.Repo
	alias Pleroma.Web.Websub.{WebsubServerSubscription, WebsubClientSubscription}
	alias Pleroma.Web.OStatus.FeedRepresenter
	alias Pleroma.Web.{XML, Endpoint, OStatus}
	alias Pleroma.Web.Router.Helpers
	require Logger

	import Ecto.Query

	@httpoison Application.get_env(:pleroma, :httpoison)

	def verify(subscription, getter \\ &@httpoison.get/3) do
	challenge = Base.encode16(:crypto.strong_rand_bytes(8))
	lease_seconds = NaiveDateTime.diff(subscription.valid_until, subscription.updated_at)
	lease_seconds = lease_seconds \|> to_string

	params = %{
	"hub.challenge": challenge,
	"hub.lease_seconds": lease_seconds,
	"hub.topic": subscription.topic,
	"hub.mode": "subscribe"
	}

	url = hd(String.split(subscription.callback, "?"))
	query = URI.parse(subscription.callback).query \|\| ""
	params = Map.merge(params, URI.decode_query(query))

	with {:ok, response} <- getter.(url, [], params: params),
	^challenge <- response.body do
	changeset = Changeset.change(subscription, %{state: "active"})
	Repo.update(changeset)
	else
	e ->
	Logger.debug("Couldn't verify subscription")
	Logger.debug(inspect(e))
	{:error, subscription}
	end
	end

	@supported_activities [
	"Create",
	"Follow",
	"Like",
	"Announce",
	"Undo",
	"Delete"
	]
	def publish(topic, user, %{data: %{"type" => type}} = activity)
	when type in @supported_activities do
	# TODO: Only send to still valid subscriptions.
	query =
	from(
	sub in WebsubServerSubscription,
	where: sub.topic == ^topic and sub.state == "active",
	where: fragment("? > NOW()", sub.valid_until)
	)

	subscriptions = Repo.all(query)

	Enum.each(subscriptions, fn sub ->
	response =
	user
	\|> FeedRepresenter.to_simple_form([activity], [user])
	\|> :xmerl.export_simple(:xmerl_xml)
	\|> to_string

	data = %{
	xml: response,
	topic: topic,
	callback: sub.callback,
	secret: sub.secret
	}

	Pleroma.Web.Federator.enqueue(:publish_single_websub, data)
	end)
	end

	def publish(_, _, _), do: ""

	def sign(secret, doc) do
	:crypto.hmac(:sha, secret, to_string(doc)) \|> Base.encode16() \|> String.downcase()
	end

	def incoming_subscription_request(user, %{"hub.mode" => "subscribe"} = params) do
	with {:ok, topic} <- valid_topic(params, user),
	{:ok, lease_time} <- lease_time(params),
	secret <- params["hub.secret"],
	callback <- params["hub.callback"] do
	subscription = get_subscription(topic, callback)

	data = %{
	state: subscription.state \|\| "requested",
	topic: topic,
	secret: secret,
	callback: callback
	}

	change = Changeset.change(subscription, data)
	websub = Repo.insert_or_update!(change)

	change =
	Changeset.change(websub, %{valid_until: NaiveDateTime.add(websub.updated_at, lease_time)})

	websub = Repo.update!(change)

	Pleroma.Web.Federator.enqueue(:verify_websub, websub)

	{:ok, websub}
	else
	{:error, reason} ->
	Logger.debug("Couldn't create subscription")
	Logger.debug(inspect(reason))

	{:error, reason}
	end
	end

	+ def incoming_subscription_request(user, params) do
	+ Logger.info("Unhandled WebSub request for #{user.nickname}: #{inspect(params)}")
	+
	+ {:error, "Invalid WebSub request"}
	+ end
	+
	defp get_subscription(topic, callback) do
	Repo.get_by(WebsubServerSubscription, topic: topic, callback: callback) \|\|
	%WebsubServerSubscription{}
	end

	# Temp hack for mastodon.
	defp lease_time(%{"hub.lease_seconds" => ""}) do
	# three days
	{:ok, 60 * 60 * 24 * 3}
	end

	defp lease_time(%{"hub.lease_seconds" => lease_seconds}) do
	{:ok, String.to_integer(lease_seconds)}
	end

	defp lease_time(_) do
	# three days
	{:ok, 60 * 60 * 24 * 3}
	end

	defp valid_topic(%{"hub.topic" => topic}, user) do
	if topic == OStatus.feed_path(user) do
	{:ok, OStatus.feed_path(user)}
	else
	{:error, "Wrong topic requested, expected #{OStatus.feed_path(user)}, got #{topic}"}
	end
	end

	def subscribe(subscriber, subscribed, requester \\ &request_subscription/1) do
	topic = subscribed.info.topic
	# FIXME: Race condition, use transactions
	{:ok, subscription} =
	with subscription when not is_nil(subscription) <-
	Repo.get_by(WebsubClientSubscription, topic: topic) do
	subscribers = [subscriber.ap_id \| subscription.subscribers] \|> Enum.uniq()
	change = Ecto.Changeset.change(subscription, %{subscribers: subscribers})
	Repo.update(change)
	else
	_e ->
	subscription = %WebsubClientSubscription{
	topic: topic,
	hub: subscribed.info.hub,
	subscribers: [subscriber.ap_id],
	state: "requested",
	secret: :crypto.strong_rand_bytes(8) \|> Base.url_encode64(),
	user: subscribed
	}

	Repo.insert(subscription)
	end

	requester.(subscription)
	end

	def gather_feed_data(topic, getter \\ &@httpoison.get/1) do
	with {:ok, response} <- getter.(topic),
	status when status in 200..299 <- response.status,
	body <- response.body,
	doc <- XML.parse_document(body),
	uri when not is_nil(uri) <- XML.string_from_xpath("/feed/author[1]/uri", doc),
	hub when not is_nil(hub) <- XML.string_from_xpath(~S{/feed/link[@rel="hub"]/@href}, doc) do
	name = XML.string_from_xpath("/feed/author[1]/name", doc)
	preferredUsername = XML.string_from_xpath("/feed/author[1]/poco:preferredUsername", doc)
	displayName = XML.string_from_xpath("/feed/author[1]/poco:displayName", doc)
	avatar = OStatus.make_avatar_object(doc)
	bio = XML.string_from_xpath("/feed/author[1]/summary", doc)

	{:ok,
	%{
	"uri" => uri,
	"hub" => hub,
	"nickname" => preferredUsername \|\| name,
	"name" => displayName \|\| name,
	"host" => URI.parse(uri).host,
	"avatar" => avatar,
	"bio" => bio
	}}
	else
	e ->
	{:error, e}
	end
	end

	def request_subscription(websub, poster \\ &@httpoison.post/3, timeout \\ 10_000) do
	data = [
	"hub.mode": "subscribe",
	"hub.topic": websub.topic,
	"hub.secret": websub.secret,
	"hub.callback": Helpers.websub_url(Endpoint, :websub_subscription_confirmation, websub.id)
	]

	# This checks once a second if we are confirmed yet
	websub_checker = fn ->
	helper = fn helper ->
	:timer.sleep(1000)
	websub = Repo.get_by(WebsubClientSubscription, id: websub.id, state: "accepted")
	if websub, do: websub, else: helper.(helper)
	end

	helper.(helper)
	end

	task = Task.async(websub_checker)

	with {:ok, %{status: 202}} <-
	poster.(websub.hub, {:form, data}, "Content-type": "application/x-www-form-urlencoded"),
	{:ok, websub} <- Task.yield(task, timeout) do
	{:ok, websub}
	else
	e ->
	Task.shutdown(task)

	change = Ecto.Changeset.change(websub, %{state: "rejected"})
	{:ok, websub} = Repo.update(change)

	Logger.debug(fn -> "Couldn't confirm subscription: #{inspect(websub)}" end)
	Logger.debug(fn -> "error: #{inspect(e)}" end)

	{:error, websub}
	end
	end

	def refresh_subscriptions(delta \\ 60 * 60 * 24) do
	Logger.debug("Refreshing subscriptions")

	cut_off = NaiveDateTime.add(NaiveDateTime.utc_now(), delta)

	query = from(sub in WebsubClientSubscription, where: sub.valid_until < ^cut_off)

	subs = Repo.all(query)

	Enum.each(subs, fn sub ->
	Pleroma.Web.Federator.enqueue(:request_subscription, sub)
	end)
	end

	def publish_one(%{xml: xml, topic: topic, callback: callback, secret: secret}) do
	signature = sign(secret \|\| "", xml)
	Logger.info(fn -> "Pushing #{topic} to #{callback}" end)

	with {:ok, %{status: code}} <-
	@httpoison.post(
	callback,
	xml,
	[
	{"Content-Type", "application/atom+xml"},
	{"X-Hub-Signature", "sha1=#{signature}"}
	]
	) do
	Logger.info(fn -> "Pushed to #{callback}, code #{code}" end)
	{:ok, code}
	else
	e ->
	Logger.debug(fn -> "Couldn't push to #{callback}, #{inspect(e)}" end)
	{:error, e}
	end
	end
	end
	diff --git a/lib/pleroma/web/websub/websub_controller.ex b/lib/pleroma/web/websub/websub_controller.ex
	index 27304d988..e58f144e5 100644
	--- a/lib/pleroma/web/websub/websub_controller.ex
	+++ b/lib/pleroma/web/websub/websub_controller.ex
	@@ -1,88 +1,95 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Websub.WebsubController do
	use Pleroma.Web, :controller
	alias Pleroma.{Repo, User}
	alias Pleroma.Web.{Websub, Federator}
	alias Pleroma.Web.Websub.WebsubClientSubscription
	require Logger

	plug(
	Pleroma.Web.FederatingPlug
	when action in [
	:websub_subscription_request,
	:websub_subscription_confirmation,
	:websub_incoming
	]
	)

	def websub_subscription_request(conn, %{"nickname" => nickname} = params) do
	user = User.get_cached_by_nickname(nickname)

	with {:ok, _websub} <- Websub.incoming_subscription_request(user, params) do
	conn
	\|> send_resp(202, "Accepted")
	else
	{:error, reason} ->
	conn
	\|> send_resp(500, reason)
	end
	end

	# TODO: Extract this into the Websub module
	def websub_subscription_confirmation(
	conn,
	%{
	"id" => id,
	"hub.mode" => "subscribe",
	"hub.challenge" => challenge,
	"hub.topic" => topic
	} = params
	) do
	Logger.debug("Got WebSub confirmation")
	Logger.debug(inspect(params))

	lease_seconds =
	if params["hub.lease_seconds"] do
	String.to_integer(params["hub.lease_seconds"])
	else
	# Guess 3 days
	60 * 60 * 24 * 3
	end

	with %WebsubClientSubscription{} = websub <-
	Repo.get_by(WebsubClientSubscription, id: id, topic: topic) do
	valid_until = NaiveDateTime.add(NaiveDateTime.utc_now(), lease_seconds)
	change = Ecto.Changeset.change(websub, %{state: "accepted", valid_until: valid_until})
	{:ok, _websub} = Repo.update(change)

	conn
	\|> send_resp(200, challenge)
	else
	_e ->
	conn
	\|> send_resp(500, "Error")
	end
	end

	+ def websub_subscription_confirmation(conn, params) do
	+ Logger.info("Invalid WebSub confirmation request: #{inspect(params)}")
	+
	+ conn
	+ \|> send_resp(500, "Invalid parameters")
	+ end
	+
	def websub_incoming(conn, %{"id" => id}) do
	with "sha1=" <> signature <- hd(get_req_header(conn, "x-hub-signature")),
	signature <- String.downcase(signature),
	%WebsubClientSubscription{} = websub <- Repo.get(WebsubClientSubscription, id),
	{:ok, body, _conn} = read_body(conn),
	^signature <- Websub.sign(websub.secret, body) do
	Federator.enqueue(:incoming_doc, body)

	conn
	\|> send_resp(200, "OK")
	else
	_e ->
	Logger.debug("Can't handle incoming subscription post")

	conn
	\|> send_resp(500, "Error")
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Sun, Nov 1, 11:11 AM (11 h, 18 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1802972
Default Alt Text: (12 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions