No OneTemporary
Actions

Size

53 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/activity.ex b/lib/pleroma/activity.ex
	index 46552c7be..0af198f7e 100644
	--- a/lib/pleroma/activity.ex
	+++ b/lib/pleroma/activity.ex
	@@ -1,348 +1,350 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Activity do
	use Ecto.Schema

	alias Pleroma.Activity
	alias Pleroma.Bookmark
	+ alias Pleroma.FederationFailure
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.Repo
	alias Pleroma.ThreadMute
	alias Pleroma.User

	import Ecto.Changeset
	import Ecto.Query

	@type t :: %__MODULE__{}
	@type actor :: String.t()

	@primary_key {:id, Pleroma.FlakeId, autogenerate: true}

	# https://github.com/tootsuite/mastodon/blob/master/app/models/notification.rb#L19
	@mastodon_notification_types %{
	"Create" => "mention",
	"Follow" => "follow",
	"Announce" => "reblog",
	"Like" => "favourite"
	}

	@mastodon_to_ap_notification_types for {k, v} <- @mastodon_notification_types,
	into: %{},
	do: {v, k}

	schema "activities" do
	field(:data, :map)
	field(:local, :boolean, default: true)
	field(:actor, :string)
	field(:recipients, {:array, :string}, default: [])
	field(:thread_muted?, :boolean, virtual: true)
	# This is a fake relation, do not use outside of with_preloaded_bookmark/get_bookmark
	has_one(:bookmark, Bookmark)
	has_many(:notifications, Notification, on_delete: :delete_all)
	+ has_many(:federation_failures, FederationFailure, on_delete: :delete_all)

	# Attention: this is a fake relation, don't try to preload it blindly and expect it to work!
	# The foreign key is embedded in a jsonb field.
	#
	# To use it, you probably want to do an inner join and a preload:
	#
	# ```
	# \|> join(:inner, [activity], o in Object,
	# on: fragment("(?->>'id') = COALESCE((?)->'object'->> 'id', (?)->>'object')",
	# o.data, activity.data, activity.data))
	# \|> preload([activity, object], [object: object])
	# ```
	#
	# As a convenience, Activity.with_preloaded_object() sets up an inner join and preload for the
	# typical case.
	has_one(:object, Object, on_delete: :nothing, foreign_key: :id)

	timestamps()
	end

	def with_joined_object(query) do
	join(query, :inner, [activity], o in Object,
	on:
	fragment(
	"(?->>'id') = COALESCE(?->'object'->>'id', ?->>'object')",
	o.data,
	activity.data,
	activity.data
	),
	as: :object
	)
	end

	def with_preloaded_object(query) do
	query
	\|> has_named_binding?(:object)
	\|> if(do: query, else: with_joined_object(query))
	\|> preload([activity, object: object], object: object)
	end

	def with_preloaded_bookmark(query, %User{} = user) do
	from([a] in query,
	left_join: b in Bookmark,
	on: b.user_id == ^user.id and b.activity_id == a.id,
	preload: [bookmark: b]
	)
	end

	def with_preloaded_bookmark(query, _), do: query

	def with_set_thread_muted_field(query, %User{} = user) do
	from([a] in query,
	left_join: tm in ThreadMute,
	on: tm.user_id == ^user.id and tm.context == fragment("?->>'context'", a.data),
	select: %Activity{a \| thread_muted?: not is_nil(tm.id)}
	)
	end

	def with_set_thread_muted_field(query, _), do: query

	def get_by_ap_id(ap_id) do
	Repo.one(
	from(
	activity in Activity,
	where: fragment("(?)->>'id' = ?", activity.data, ^to_string(ap_id))
	)
	)
	end

	def get_bookmark(%Activity{} = activity, %User{} = user) do
	if Ecto.assoc_loaded?(activity.bookmark) do
	activity.bookmark
	else
	Bookmark.get(user.id, activity.id)
	end
	end

	def get_bookmark(_, _), do: nil

	def change(struct, params \\ %{}) do
	struct
	\|> cast(params, [:data, :recipients])
	\|> validate_required([:data])
	\|> unique_constraint(:ap_id, name: :activities_unique_apid_index)
	end

	def get_by_ap_id_with_object(ap_id) do
	Repo.one(
	from(
	activity in Activity,
	where: fragment("(?)->>'id' = ?", activity.data, ^to_string(ap_id)),
	left_join: o in Object,
	on:
	fragment(
	"(?->>'id') = COALESCE(?->'object'->>'id', ?->>'object')",
	o.data,
	activity.data,
	activity.data
	),
	preload: [object: o]
	)
	)
	end

	def get_by_id(id) do
	Activity
	\|> where([a], a.id == ^id)
	\|> restrict_deactivated_users()
	\|> Repo.one()
	end

	def get_by_id_with_object(id) do
	from(activity in Activity,
	where: activity.id == ^id,
	inner_join: o in Object,
	on:
	fragment(
	"(?->>'id') = COALESCE(?->'object'->>'id', ?->>'object')",
	o.data,
	activity.data,
	activity.data
	),
	preload: [object: o]
	)
	\|> Repo.one()
	end

	def by_object_ap_id(ap_id) do
	from(
	activity in Activity,
	where:
	fragment(
	"coalesce((?)->'object'->>'id', (?)->>'object') = ?",
	activity.data,
	activity.data,
	^to_string(ap_id)
	)
	)
	end

	def create_by_object_ap_id(ap_ids) when is_list(ap_ids) do
	from(
	activity in Activity,
	where:
	fragment(
	"coalesce((?)->'object'->>'id', (?)->>'object') = ANY(?)",
	activity.data,
	activity.data,
	^ap_ids
	),
	where: fragment("(?)->>'type' = 'Create'", activity.data)
	)
	end

	def create_by_object_ap_id(ap_id) when is_binary(ap_id) do
	from(
	activity in Activity,
	where:
	fragment(
	"coalesce((?)->'object'->>'id', (?)->>'object') = ?",
	activity.data,
	activity.data,
	^to_string(ap_id)
	),
	where: fragment("(?)->>'type' = 'Create'", activity.data)
	)
	end

	def create_by_object_ap_id(_), do: nil

	def get_all_create_by_object_ap_id(ap_id) do
	Repo.all(create_by_object_ap_id(ap_id))
	end

	def get_create_by_object_ap_id(ap_id) when is_binary(ap_id) do
	create_by_object_ap_id(ap_id)
	\|> restrict_deactivated_users()
	\|> Repo.one()
	end

	def get_create_by_object_ap_id(_), do: nil

	def create_by_object_ap_id_with_object(ap_id) when is_binary(ap_id) do
	from(
	activity in Activity,
	where:
	fragment(
	"coalesce((?)->'object'->>'id', (?)->>'object') = ?",
	activity.data,
	activity.data,
	^to_string(ap_id)
	),
	where: fragment("(?)->>'type' = 'Create'", activity.data),
	inner_join: o in Object,
	on:
	fragment(
	"(?->>'id') = COALESCE(?->'object'->>'id', ?->>'object')",
	o.data,
	activity.data,
	activity.data
	),
	preload: [object: o]
	)
	end

	def create_by_object_ap_id_with_object(_), do: nil

	def get_create_by_object_ap_id_with_object(ap_id) when is_binary(ap_id) do
	ap_id
	\|> create_by_object_ap_id_with_object()
	\|> Repo.one()
	end

	def get_create_by_object_ap_id_with_object(_), do: nil

	defp get_in_reply_to_activity_from_object(%Object{data: %{"inReplyTo" => ap_id}}) do
	get_create_by_object_ap_id_with_object(ap_id)
	end

	defp get_in_reply_to_activity_from_object(_), do: nil

	def get_in_reply_to_activity(%Activity{data: %{"object" => object}}) do
	get_in_reply_to_activity_from_object(Object.normalize(object))
	end

	def normalize(obj) when is_map(obj), do: get_by_ap_id_with_object(obj["id"])
	def normalize(ap_id) when is_binary(ap_id), do: get_by_ap_id_with_object(ap_id)
	def normalize(_), do: nil

	def delete_by_ap_id(id) when is_binary(id) do
	by_object_ap_id(id)
	\|> select([u], u)
	\|> Repo.delete_all()
	\|> elem(1)
	\|> Enum.find(fn
	%{data: %{"type" => "Create", "object" => ap_id}} when is_binary(ap_id) -> ap_id == id
	%{data: %{"type" => "Create", "object" => %{"id" => ap_id}}} -> ap_id == id
	_ -> nil
	end)
	end

	def delete_by_ap_id(_), do: nil

	for {ap_type, type} <- @mastodon_notification_types do
	def mastodon_notification_type(%Activity{data: %{"type" => unquote(ap_type)}}),
	do: unquote(type)
	end

	def mastodon_notification_type(%Activity{}), do: nil

	def from_mastodon_notification_type(type) do
	Map.get(@mastodon_to_ap_notification_types, type)
	end

	def all_by_actor_and_id(actor, status_ids \\ [])
	def all_by_actor_and_id(_actor, []), do: []

	def all_by_actor_and_id(actor, status_ids) do
	Activity
	\|> where([s], s.id in ^status_ids)
	\|> where([s], s.actor == ^actor)
	\|> Repo.all()
	end

	def follow_requests_for_actor(%Pleroma.User{ap_id: ap_id}) do
	from(
	a in Activity,
	where:
	fragment(
	"? ->> 'type' = 'Follow'",
	a.data
	),
	where:
	fragment(
	"? ->> 'state' = 'pending'",
	a.data
	),
	where:
	fragment(
	"coalesce((?)->'object'->>'id', (?)->>'object') = ?",
	a.data,
	a.data,
	^ap_id
	)
	)
	end

	@spec query_by_actor(actor()) :: Ecto.Query.t()
	def query_by_actor(actor) do
	from(a in Activity, where: a.actor == ^actor)
	end

	def restrict_deactivated_users(query) do
	from(activity in query,
	where:
	fragment(
	"? not in (SELECT ap_id FROM users WHERE info->'deactivated' @> 'true')",
	activity.actor
	)
	)
	end

	defdelegate search(user, query, options \\ []), to: Pleroma.Activity.Search
	end
	diff --git a/lib/pleroma/federation_failure.ex b/lib/pleroma/federation_failure.ex
	new file mode 100644
	index 000000000..bda1dfbab
	--- /dev/null
	+++ b/lib/pleroma/federation_failure.ex
	@@ -0,0 +1,60 @@
	+# Pleroma: A lightweight social networking server
	+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	+# SPDX-License-Identifier: AGPL-3.0-only
	+
	+defmodule Pleroma.FederationFailure do
	+ use Ecto.Schema
	+
	+ alias Pleroma.Activity
	+ alias Pleroma.FederationFailure
	+ alias Pleroma.FlakeId
	+ alias Pleroma.Repo
	+
	+ import Ecto.Changeset
	+ import Ecto.Query
	+
	+ @type t :: %__MODULE__{}
	+
	+ schema "federation_failures" do
	+ field(:recipient, :string)
	+ field(:transport, :string)
	+ field(:data, :map)
	+ field(:retries_count, :integer, default: 0)
	+
	+ belongs_to(:activity, Activity, type: FlakeId)
	+
	+ timestamps()
	+ end
	+
	+ def changeset(federation_failure, params \\ %{}) do
	+ federation_failure
	+ \|> cast(params, [:activity_id, :recipient, :transport, :data, :retries_count])
	+ \|> validate_required([:activity_id, :recipient, :transport, :data])
	+ \|> foreign_key_constraint(:activity_id)
	+ \|> unique_constraint(:activity_id,
	+ name: :federation_failures_activity_id_recipient_transport_index
	+ )
	+ end
	+
	+ def create_or_rewrite_with(%{activity_id: nil}), do: :noop
	+
	+ def create_or_rewrite_with(%{} = params) do
	+ %FederationFailure{}
	+ \|> changeset(params)
	+ \|> Repo.insert(
	+ on_conflict: :replace_all_except_primary_key,
	+ conflict_target: [:activity_id, :recipient, :transport]
	+ )
	+ end
	+
	+ def delete_by(%{activity_id: nil}), do: :noop
	+
	+ def delete_by(%{activity_id: activity_id, recipient: recipient, transport: transport}) do
	+ from(ff in FederationFailure,
	+ where:
	+ ff.activity_id == ^activity_id and ff.recipient == ^recipient and
	+ ff.transport == ^transport
	+ )
	+ \|> Repo.delete_all()
	+ end
	+end
	diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex
	index c505223f7..de9492319 100644
	--- a/lib/pleroma/web/activity_pub/publisher.ex
	+++ b/lib/pleroma/web/activity_pub/publisher.ex
	@@ -1,201 +1,209 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.Publisher do
	alias Pleroma.Activity
	alias Pleroma.Config
	alias Pleroma.HTTP
	alias Pleroma.Instances
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.Relay
	alias Pleroma.Web.ActivityPub.Transmogrifier

	import Pleroma.Web.ActivityPub.Visibility

	@behaviour Pleroma.Web.Federator.Publisher

	require Logger

	@moduledoc """
	ActivityPub outgoing federation module.
	"""

	@doc """
	Determine if an activity can be represented by running it through Transmogrifier.
	"""
	def is_representable?(%Activity{} = activity) do
	with {:ok, _data} <- Transmogrifier.prepare_outgoing(activity.data) do
	true
	else
	_e ->
	false
	end
	end

	@doc """
	Publish a single message to a peer. Takes a struct with the following
	parameters set:

	* `inbox`: the inbox to publish to
	* `json`: the JSON message body representing the ActivityPub message
	* `actor`: the actor which is signing the message
	* `id`: the ActivityStreams URI of the message
	"""
	def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = params) do
	Logger.info("Federating #{id} to #{inbox}")
	host = URI.parse(inbox).host

	digest = "SHA-256=" <> (:crypto.hash(:sha256, json) \|> Base.encode64())

	date =
	NaiveDateTime.utc_now()
	\|> Timex.format!("{WDshort}, {0D} {Mshort} {YYYY} {h24}:{m}:{s} GMT")

	signature =
	Pleroma.Signature.sign(actor, %{
	host: host,
	"content-length": byte_size(json),
	digest: digest,
	date: date
	})

	with {:ok, %{status: code}} when code in 200..299 <-
	result =
	HTTP.post(
	inbox,
	json,
	[
	{"Content-Type", "application/activity+json"},
	{"Date", date},
	{"signature", signature},
	{"digest", digest}
	]
	) do
	if !Map.has_key?(params, :unreachable_since) \|\| params[:unreachable_since],
	do: Instances.set_reachable(inbox)

	result
	else
	{_post_result, response} ->
	unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
	{:error, response}
	end
	end

	defp should_federate?(inbox, public) do
	if public do
	true
	else
	inbox_info = URI.parse(inbox)
	!Enum.member?(Config.get([:instance, :quarantined_instances], []), inbox_info.host)
	end
	end

	defp recipients(actor, activity) do
	followers =
	if actor.follower_address in activity.recipients do
	{:ok, followers} = User.get_followers(actor)
	Enum.filter(followers, &(!&1.local))
	else
	[]
	end

	Pleroma.Web.Salmon.remote_users(actor, activity) ++ followers
	end

	defp get_cc_ap_ids(ap_id, recipients) do
	host = Map.get(URI.parse(ap_id), :host)

	recipients
	\|> Enum.filter(fn %User{ap_id: ap_id} -> Map.get(URI.parse(ap_id), :host) == host end)
	\|> Enum.map(& &1.ap_id)
	end

	@doc """
	Publishes an activity with BCC to all relevant peers.
	"""

	def publish(actor, %{data: %{"bcc" => bcc}} = activity) when is_list(bcc) and bcc != [] do
	public = is_public?(activity)
	{:ok, data} = Transmogrifier.prepare_outgoing(activity.data)

	recipients = recipients(actor, activity)

	recipients
	\|> Enum.filter(&User.ap_enabled?/1)
	\|> Enum.map(fn %{info: %{source_data: data}} -> data["inbox"] end)
	\|> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
	\|> Instances.filter_reachable()
	\|> Enum.each(fn {inbox, unreachable_since} ->
	%User{ap_id: ap_id} =
	Enum.find(recipients, fn %{info: %{source_data: data}} -> data["inbox"] == inbox end)

	# Get all the recipients on the same host and add them to cc. Otherwise, a remote
	# instance would only accept a first message for the first recipient and ignore the rest.
	cc = get_cc_ap_ids(ap_id, recipients)

	json =
	data
	\|> Map.put("cc", cc)
	\|> Jason.encode!()

	Pleroma.Web.Federator.Publisher.enqueue_one(__MODULE__, %{
	inbox: inbox,
	json: json,
	actor: actor,
	id: activity.data["id"],
	- unreachable_since: unreachable_since
	+ unreachable_since: unreachable_since,
	+ job: %{
	+ activity_id: activity.id,
	+ recipient: inbox
	+ }
	})
	end)
	end

	@doc """
	Publishes an activity to all relevant peers.
	"""
	def publish(%User{} = actor, %Activity{} = activity) do
	public = is_public?(activity)

	if public && Config.get([:instance, :allow_relay]) do
	Logger.info(fn -> "Relaying #{activity.data["id"]} out" end)
	Relay.publish(activity)
	end

	{:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
	json = Jason.encode!(data)

	recipients(actor, activity)
	\|> Enum.filter(fn user -> User.ap_enabled?(user) end)
	\|> Enum.map(fn %{info: %{source_data: data}} ->
	(is_map(data["endpoints"]) && Map.get(data["endpoints"], "sharedInbox")) \|\| data["inbox"]
	end)
	\|> Enum.uniq()
	\|> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
	\|> Instances.filter_reachable()
	\|> Enum.each(fn {inbox, unreachable_since} ->
	Pleroma.Web.Federator.Publisher.enqueue_one(
	__MODULE__,
	%{
	inbox: inbox,
	json: json,
	actor: actor,
	id: activity.data["id"],
	- unreachable_since: unreachable_since
	+ unreachable_since: unreachable_since,
	+ job: %{
	+ activity_id: activity.id,
	+ recipient: inbox
	+ }
	}
	)
	end)
	end

	def gather_webfinger_links(%User{} = user) do
	[
	%{"rel" => "self", "type" => "application/activity+json", "href" => user.ap_id},
	%{
	"rel" => "self",
	"type" => "application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"",
	"href" => user.ap_id
	}
	]
	end

	def gather_nodeinfo_protocol_names, do: ["activitypub"]
	end
	diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex
	index f4f9e83e0..97ec9d549 100644
	--- a/lib/pleroma/web/federator/federator.ex
	+++ b/lib/pleroma/web/federator/federator.ex
	@@ -1,160 +1,146 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Federator do
	alias Pleroma.Activity
	alias Pleroma.Object.Containment
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.ActivityPub
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.ActivityPub.Utils
	alias Pleroma.Web.Federator.Publisher
	- alias Pleroma.Web.Federator.RetryQueue
	alias Pleroma.Web.OStatus
	alias Pleroma.Web.Websub

	require Logger

	def init do
	# 1 minute
	Process.sleep(1000 * 60)
	refresh_subscriptions()
	end

	@doc "Addresses [memory leaks on recursive replies fetching](https://git.pleroma.social/pleroma/pleroma/issues/161)"
	# credo:disable-for-previous-line Credo.Check.Readability.MaxLineLength
	def allowed_incoming_reply_depth?(depth) do
	max_replies_depth = Pleroma.Config.get([:instance, :federation_incoming_replies_max_depth])

	if max_replies_depth do
	(depth \|\| 1) <= max_replies_depth
	else
	true
	end
	end

	# Client API

	def incoming_doc(doc) do
	PleromaJobQueue.enqueue(:federator_incoming, __MODULE__, [:incoming_doc, doc])
	end

	def incoming_ap_doc(params) do
	PleromaJobQueue.enqueue(:federator_incoming, __MODULE__, [:incoming_ap_doc, params])
	end

	def publish(activity, priority \\ 1) do
	PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish, activity], priority)
	end

	def verify_websub(websub) do
	PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:verify_websub, websub])
	end

	def request_subscription(sub) do
	PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:request_subscription, sub])
	end

	def refresh_subscriptions do
	PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:refresh_subscriptions])
	end

	# Job Worker Callbacks

	def perform(:refresh_subscriptions) do
	Logger.debug("Federator running refresh subscriptions")
	Websub.refresh_subscriptions()

	spawn(fn ->
	# 6 hours
	Process.sleep(1000 * 60 * 60 * 6)
	refresh_subscriptions()
	end)
	end

	def perform(:request_subscription, websub) do
	Logger.debug("Refreshing #{websub.topic}")

	with {:ok, websub} <- Websub.request_subscription(websub) do
	Logger.debug("Successfully refreshed #{websub.topic}")
	else
	_e -> Logger.debug("Couldn't refresh #{websub.topic}")
	end
	end

	def perform(:publish, activity) do
	Logger.debug(fn -> "Running publish for #{activity.data["id"]}" end)

	with %User{} = actor <- User.get_cached_by_ap_id(activity.data["actor"]),
	{:ok, actor} <- User.ensure_keys_present(actor) do
	Publisher.publish(actor, activity)
	end
	end

	def perform(:verify_websub, websub) do
	Logger.debug(fn ->
	"Running WebSub verification for #{websub.id} (#{websub.topic}, #{websub.callback})"
	end)

	Websub.verify(websub)
	end

	def perform(:incoming_doc, doc) do
	Logger.info("Got document, trying to parse")
	OStatus.handle_incoming(doc)
	end

	def perform(:incoming_ap_doc, params) do
	Logger.info("Handling incoming AP activity")

	params = Utils.normalize_params(params)

	# NOTE: we use the actor ID to do the containment, this is fine because an
	# actor shouldn't be acting on objects outside their own AP server.
	with {:ok, _user} <- ap_enabled_actor(params["actor"]),
	nil <- Activity.normalize(params["id"]),
	:ok <- Containment.contain_origin_from_id(params["actor"], params),
	{:ok, activity} <- Transmogrifier.handle_incoming(params) do
	{:ok, activity}
	else
	%Activity{} ->
	Logger.info("Already had #{params["id"]}")
	:error

	_e ->
	# Just drop those for now
	Logger.info("Unhandled activity")
	Logger.info(Jason.encode!(params, pretty: true))
	:error
	end
	end

	- def perform(
	- :publish_single_websub,
	- %{xml: _xml, topic: _topic, callback: _callback, secret: _secret} = params
	- ) do
	- case Websub.publish_one(params) do
	- {:ok, _} ->
	- :ok
	-
	- {:error, _} ->
	- RetryQueue.enqueue(params, Websub)
	- end
	- end
	-
	def perform(type, _) do
	Logger.debug(fn -> "Unknown task: #{type}" end)
	{:error, "Don't know what to do with this"}
	end

	def ap_enabled_actor(id) do
	user = User.get_cached_by_ap_id(id)

	if User.ap_enabled?(user) do
	{:ok, user}
	else
	ActivityPub.make_user_from_ap_id(id)
	end
	end
	end
	diff --git a/lib/pleroma/web/federator/publisher.ex b/lib/pleroma/web/federator/publisher.ex
	index 70f870244..6f0a03814 100644
	--- a/lib/pleroma/web/federator/publisher.ex
	+++ b/lib/pleroma/web/federator/publisher.ex
	@@ -1,95 +1,111 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Federator.Publisher do
	alias Pleroma.Activity
	alias Pleroma.Config
	+ alias Pleroma.FederationFailure
	alias Pleroma.User
	alias Pleroma.Web.Federator.RetryQueue

	require Logger

	@moduledoc """
	Defines the contract used by federation implementations to publish messages to
	their peers.
	"""

	@doc """
	Determine whether an activity can be relayed using the federation module.
	"""
	@callback is_representable?(Pleroma.Activity.t()) :: boolean()

	@doc """
	Relays an activity to a specified peer, determined by the parameters. The
	parameters used are controlled by the federation module.
	"""
	@callback publish_one(Map.t()) :: {:ok, Map.t()} \| {:error, any()}

	@doc """
	Enqueue publishing a single activity.
	"""
	@spec enqueue_one(module(), Map.t()) :: :ok
	- def enqueue_one(module, %{} = params),
	- do: PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_one, module, params])
	+ def enqueue_one(transport, %{job: %{activity_id: _, recipient: _}} = params) do
	+ PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_one, transport, params])
	+ end

	@spec perform(atom(), module(), any()) :: {:ok, any()} \| {:error, any()}
	- def perform(:publish_one, module, params) do
	- case apply(module, :publish_one, [params]) do
	+ def perform(
	+ :publish_one,
	+ transport,
	+ %{job: %{activity_id: _, recipient: _}} = params
	+ ) do
	+ federation_failure_id_data = Map.put(params[:job], :transport, to_string(transport))
	+
	+ case apply(transport, :publish_one, [params]) do
	{:ok, _} ->
	+ FederationFailure.delete_by(federation_failure_id_data)
	:ok

	{:error, _e} ->
	- RetryQueue.enqueue(params, module)
	+ federation_failure_id_data
	+ # Note: `params` must be JSON-serializable (it's not for Salmon, see salmon_test.exs)
	+ # \|> Map.put(:data, params)
	+ \|> Map.put(:data, %{})
	+ \|> Map.put(:retries_count, 0)
	+ \|> FederationFailure.create_or_rewrite_with()
	+
	+ RetryQueue.enqueue(params, transport)
	end
	end

	def perform(type, _, _) do
	Logger.debug("Unknown task: #{type}")
	{:error, "Don't know what to do with this"}
	end

	@doc """
	Relays an activity to all specified peers.
	"""
	@callback publish(User.t(), Activity.t()) :: :ok \| {:error, any()}

	@spec publish(User.t(), Activity.t()) :: :ok
	def publish(%User{} = user, %Activity{} = activity) do
	Config.get([:instance, :federation_publisher_modules])
	\|> Enum.each(fn module ->
	if module.is_representable?(activity) do
	Logger.info("Publishing #{activity.data["id"]} using #{inspect(module)}")
	module.publish(user, activity)
	end
	end)

	:ok
	end

	@doc """
	Gathers links used by an outgoing federation module for WebFinger output.
	"""
	@callback gather_webfinger_links(User.t()) :: list()

	@spec gather_webfinger_links(User.t()) :: list()
	def gather_webfinger_links(%User{} = user) do
	Config.get([:instance, :federation_publisher_modules])
	\|> Enum.reduce([], fn module, links ->
	links ++ module.gather_webfinger_links(user)
	end)
	end

	@doc """
	Gathers nodeinfo protocol names supported by the federation module.
	"""
	@callback gather_nodeinfo_protocol_names() :: list()

	@spec gather_nodeinfo_protocol_names() :: list()
	def gather_nodeinfo_protocol_names do
	Config.get([:instance, :federation_publisher_modules])
	\|> Enum.reduce([], fn module, links ->
	links ++ module.gather_nodeinfo_protocol_names()
	end)
	end
	end
	diff --git a/lib/pleroma/web/federator/retry_queue.ex b/lib/pleroma/web/federator/retry_queue.ex
	index 3db948c2e..cba7f72e7 100644
	--- a/lib/pleroma/web/federator/retry_queue.ex
	+++ b/lib/pleroma/web/federator/retry_queue.ex
	@@ -1,239 +1,258 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Federator.RetryQueue do
	use GenServer

	+ alias Pleroma.FederationFailure
	+
	require Logger

	def init(args) do
	queue_table = :ets.new(:pleroma_retry_queue, [:bag, :protected])

	{:ok, %{args \| queue_table: queue_table, running_jobs: :sets.new()}}
	end

	def start_link do
	enabled =
	if Pleroma.Config.get(:env) == :test,
	do: true,
	else: Pleroma.Config.get([__MODULE__, :enabled], false)

	if enabled do
	Logger.info("Starting retry queue")

	linkres =
	GenServer.start_link(
	__MODULE__,
	%{delivered: 0, dropped: 0, queue_table: nil, running_jobs: nil},
	name: __MODULE__
	)

	maybe_kickoff_timer()
	linkres
	else
	Logger.info("Retry queue disabled")
	:ignore
	end
	end

	def enqueue(data, transport, retries \\ 0) do
	+ if job_params = job_params(data, transport) do
	+ job_params
	+ # Note: `params` must be JSON-serializable (it's not for Salmon, see salmon_test.exs)
	+ # \|> Map.put(:data, data)
	+ \|> Map.put(:data, %{})
	+ \|> Map.put(:retries_count, retries + 1)
	+ \|> FederationFailure.create_or_rewrite_with()
	+ end
	+
	GenServer.cast(__MODULE__, {:maybe_enqueue, data, transport, retries + 1})
	end

	def get_stats do
	GenServer.call(__MODULE__, :get_stats)
	end

	def reset_stats do
	GenServer.call(__MODULE__, :reset_stats)
	end

	def get_retry_params(retries) do
	if retries > Pleroma.Config.get([__MODULE__, :max_retries]) do
	{:drop, "Max retries reached"}
	else
	{:retry, growth_function(retries)}
	end
	end

	def get_retry_timer_interval do
	Pleroma.Config.get([:retry_queue, :interval], 1000)
	end

	defp ets_count_expires(table, current_time) do
	:ets.select_count(
	table,
	[
	{
	{:"$1", :"$2"},
	[{:"=<", :"$1", {:const, current_time}}],
	[true]
	}
	]
	)
	end

	defp ets_pop_n_expired(table, current_time, desired) do
	{popped, _continuation} =
	:ets.select(
	table,
	[
	{
	{:"$1", :"$2"},
	[{:"=<", :"$1", {:const, current_time}}],
	[:"$_"]
	}
	],
	desired
	)

	popped
	\|> Enum.each(fn e ->
	:ets.delete_object(table, e)
	end)

	popped
	end

	def maybe_start_job(running_jobs, queue_table) do
	# we don't want to hit the ets or the DateTime more times than we have to
	# could optimize slightly further by not using the count, and instead grabbing
	# up to N objects early...
	current_time = DateTime.to_unix(DateTime.utc_now())
	n_running_jobs = :sets.size(running_jobs)

	if n_running_jobs < Pleroma.Config.get([__MODULE__, :max_jobs]) do
	n_ready_jobs = ets_count_expires(queue_table, current_time)

	if n_ready_jobs > 0 do
	# figure out how many we could start
	available_job_slots = Pleroma.Config.get([__MODULE__, :max_jobs]) - n_running_jobs
	start_n_jobs(running_jobs, queue_table, current_time, available_job_slots)
	else
	running_jobs
	end
	else
	running_jobs
	end
	end

	defp start_n_jobs(running_jobs, _queue_table, _current_time, 0) do
	running_jobs
	end

	defp start_n_jobs(running_jobs, queue_table, current_time, available_job_slots)
	when available_job_slots > 0 do
	candidates = ets_pop_n_expired(queue_table, current_time, available_job_slots)

	candidates
	\|> List.foldl(running_jobs, fn {_, e}, rj ->
	{:ok, pid} = Task.start(fn -> worker(e) end)
	mref = Process.monitor(pid)
	:sets.add_element(mref, rj)
	end)
	end

	def worker({:send, data, transport, retries}) do
	case transport.publish_one(data) do
	{:ok, _} ->
	+ if job_params = job_params(data, transport), do: FederationFailure.delete_by(job_params)
	GenServer.cast(__MODULE__, :inc_delivered)
	:delivered

	{:error, _reason} ->
	enqueue(data, transport, retries)
	:retry
	end
	end

	def handle_call(:get_stats, _from, %{delivered: delivery_count, dropped: drop_count} = state) do
	{:reply, %{delivered: delivery_count, dropped: drop_count}, state}
	end

	def handle_call(:reset_stats, _from, %{delivered: delivery_count, dropped: drop_count} = state) do
	{:reply, %{delivered: delivery_count, dropped: drop_count},
	%{state \| delivered: 0, dropped: 0}}
	end

	def handle_cast(:reset_stats, state) do
	{:noreply, %{state \| delivered: 0, dropped: 0}}
	end

	def handle_cast(
	{:maybe_enqueue, data, transport, retries},
	%{dropped: drop_count, queue_table: queue_table, running_jobs: running_jobs} = state
	) do
	case get_retry_params(retries) do
	{:retry, timeout} ->
	:ets.insert(queue_table, {timeout, {:send, data, transport, retries}})
	running_jobs = maybe_start_job(running_jobs, queue_table)
	{:noreply, %{state \| running_jobs: running_jobs}}

	{:drop, message} ->
	Logger.debug(message)
	{:noreply, %{state \| dropped: drop_count + 1}}
	end
	end

	def handle_cast(:kickoff_timer, state) do
	retry_interval = get_retry_timer_interval()
	Process.send_after(__MODULE__, :retry_timer_run, retry_interval)
	{:noreply, state}
	end

	def handle_cast(:inc_delivered, %{delivered: delivery_count} = state) do
	{:noreply, %{state \| delivered: delivery_count + 1}}
	end

	def handle_cast(:inc_dropped, %{dropped: drop_count} = state) do
	{:noreply, %{state \| dropped: drop_count + 1}}
	end

	def handle_info({:send, data, transport, retries}, %{delivered: delivery_count} = state) do
	case transport.publish_one(data) do
	{:ok, _} ->
	+ if job_params = job_params(data, transport), do: FederationFailure.delete_by(job_params)
	{:noreply, %{state \| delivered: delivery_count + 1}}

	{:error, _reason} ->
	enqueue(data, transport, retries)
	{:noreply, state}
	end
	end

	def handle_info(
	:retry_timer_run,
	%{queue_table: queue_table, running_jobs: running_jobs} = state
	) do
	maybe_kickoff_timer()
	running_jobs = maybe_start_job(running_jobs, queue_table)
	{:noreply, %{state \| running_jobs: running_jobs}}
	end

	def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
	%{running_jobs: running_jobs, queue_table: queue_table} = state
	running_jobs = :sets.del_element(ref, running_jobs)
	running_jobs = maybe_start_job(running_jobs, queue_table)
	{:noreply, %{state \| running_jobs: running_jobs}}
	end

	def handle_info(unknown, state) do
	Logger.debug("RetryQueue: don't know what to do with #{inspect(unknown)}, ignoring")
	{:noreply, state}
	end

	if Pleroma.Config.get(:env) == :test do
	defp growth_function(_retries) do
	_shutit = Pleroma.Config.get([__MODULE__, :initial_timeout])
	DateTime.to_unix(DateTime.utc_now()) - 1
	end
	else
	defp growth_function(retries) do
	round(Pleroma.Config.get([__MODULE__, :initial_timeout]) * :math.pow(retries, 3)) +
	DateTime.to_unix(DateTime.utc_now())
	end
	end

	defp maybe_kickoff_timer do
	GenServer.cast(__MODULE__, :kickoff_timer)
	end
	+
	+ defp job_params(%{job: job_data}, transport) do
	+ Map.put(job_data, :transport, to_string(transport))
	+ end
	+
	+ defp job_params(_, _), do: nil
	end
	diff --git a/lib/pleroma/web/salmon/salmon.ex b/lib/pleroma/web/salmon/salmon.ex
	index 9b01ebcc6..db3f2f08b 100644
	--- a/lib/pleroma/web/salmon/salmon.ex
	+++ b/lib/pleroma/web/salmon/salmon.ex
	@@ -1,245 +1,250 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Salmon do
	@behaviour Pleroma.Web.Federator.Publisher

	use Bitwise

	alias Pleroma.Activity
	alias Pleroma.HTTP
	alias Pleroma.Instances
	alias Pleroma.Keys
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.Visibility
	alias Pleroma.Web.Federator.Publisher
	alias Pleroma.Web.OStatus
	alias Pleroma.Web.OStatus.ActivityRepresenter
	alias Pleroma.Web.XML

	require Logger

	def decode(salmon) do
	doc = XML.parse_document(salmon)

	{:xmlObj, :string, data} = :xmerl_xpath.string('string(//me:data[1])', doc)
	{:xmlObj, :string, sig} = :xmerl_xpath.string('string(//me:sig[1])', doc)
	{:xmlObj, :string, alg} = :xmerl_xpath.string('string(//me:alg[1])', doc)
	{:xmlObj, :string, encoding} = :xmerl_xpath.string('string(//me:encoding[1])', doc)
	{:xmlObj, :string, type} = :xmerl_xpath.string('string(//me:data[1]/@type)', doc)

	{:ok, data} = Base.url_decode64(to_string(data), ignore: :whitespace)
	{:ok, sig} = Base.url_decode64(to_string(sig), ignore: :whitespace)
	alg = to_string(alg)
	encoding = to_string(encoding)
	type = to_string(type)

	[data, type, encoding, alg, sig]
	end

	def fetch_magic_key(salmon) do
	with [data, _, _, _, _] <- decode(salmon),
	doc <- XML.parse_document(data),
	uri when not is_nil(uri) <- XML.string_from_xpath("/entry/author[1]/uri", doc),
	{:ok, public_key} <- User.get_public_key_for_ap_id(uri),
	magic_key <- encode_key(public_key) do
	{:ok, magic_key}
	end
	end

	def decode_and_validate(magickey, salmon) do
	[data, type, encoding, alg, sig] = decode(salmon)

	signed_text =
	[data, type, encoding, alg]
	\|> Enum.map(&Base.url_encode64/1)
	\|> Enum.join(".")

	key = decode_key(magickey)

	verify = :public_key.verify(signed_text, :sha256, sig, key)

	if verify do
	{:ok, data}
	else
	:error
	end
	end

	def decode_key("RSA." <> magickey) do
	make_integer = fn bin ->
	list = :erlang.binary_to_list(bin)
	Enum.reduce(list, 0, fn el, acc -> acc <<< 8 \|\|\| el end)
	end

	[modulus, exponent] =
	magickey
	\|> String.split(".")
	\|> Enum.map(fn n -> Base.url_decode64!(n, padding: false) end)
	\|> Enum.map(make_integer)

	{:RSAPublicKey, modulus, exponent}
	end

	def encode_key({:RSAPublicKey, modulus, exponent}) do
	modulus_enc = :binary.encode_unsigned(modulus) \|> Base.url_encode64()
	exponent_enc = :binary.encode_unsigned(exponent) \|> Base.url_encode64()

	"RSA.#{modulus_enc}.#{exponent_enc}"
	end

	def encode(private_key, doc) do
	type = "application/atom+xml"
	encoding = "base64url"
	alg = "RSA-SHA256"

	signed_text =
	[doc, type, encoding, alg]
	\|> Enum.map(&Base.url_encode64/1)
	\|> Enum.join(".")

	signature =
	signed_text
	\|> :public_key.sign(:sha256, private_key)
	\|> to_string
	\|> Base.url_encode64()

	doc_base64 =
	doc
	\|> Base.url_encode64()

	# Don't need proper xml building, these strings are safe to leave unescaped
	salmon = """
	<?xml version="1.0" encoding="UTF-8"?>
	<me:env xmlns:me="http://salmon-protocol.org/ns/magic-env">
	<me:data type="application/atom+xml">#{doc_base64}</me:data>
	<me:encoding>#{encoding}</me:encoding>
	<me:alg>#{alg}</me:alg>
	<me:sig>#{signature}</me:sig>
	</me:env>
	"""

	{:ok, salmon}
	end

	def remote_users(%User{id: user_id}, %{data: %{"to" => to} = data}) do
	cc = Map.get(data, "cc", [])

	bcc =
	data
	\|> Map.get("bcc", [])
	\|> Enum.reduce([], fn ap_id, bcc ->
	case Pleroma.List.get_by_ap_id(ap_id) do
	%Pleroma.List{user_id: ^user_id} = list ->
	{:ok, following} = Pleroma.List.get_following(list)
	bcc ++ Enum.map(following, & &1.ap_id)

	_ ->
	bcc
	end
	end)

	[to, cc, bcc]
	\|> Enum.concat()
	\|> Enum.map(&User.get_cached_by_ap_id/1)
	\|> Enum.filter(fn user -> user && !user.local end)
	end

	@doc "Pushes an activity to remote account."
	def publish_one(%{recipient: %{info: %{salmon: salmon}}} = params),
	do: publish_one(Map.put(params, :recipient, salmon))

	def publish_one(%{recipient: url, feed: feed} = params) when is_binary(url) do
	with {:ok, %{status: code}} when code in 200..299 <-
	HTTP.post(
	url,
	feed,
	[{"Content-Type", "application/magic-envelope+xml"}]
	) do
	if !Map.has_key?(params, :unreachable_since) \|\| params[:unreachable_since],
	do: Instances.set_reachable(url)

	Logger.debug(fn -> "Pushed to #{url}, code #{code}" end)
	{:ok, code}
	else
	e ->
	unless params[:unreachable_since], do: Instances.set_reachable(url)
	Logger.debug(fn -> "Pushing Salmon to #{url} failed, #{inspect(e)}" end)
	{:error, "Unreachable instance"}
	end
	end

	def publish_one(_), do: :noop

	@supported_activities [
	"Create",
	"Follow",
	"Like",
	"Announce",
	"Undo",
	"Delete"
	]

	def is_representable?(%Activity{data: %{"type" => type}} = activity)
	when type in @supported_activities,
	do: Visibility.is_public?(activity)

	def is_representable?(_), do: false

	@doc """
	Publishes an activity to remote accounts
	"""
	@spec publish(User.t(), Pleroma.Activity.t()) :: none
	def publish(user, activity)

	def publish(%{info: %{keys: keys}} = user, %{data: %{"type" => type}} = activity)
	when type in @supported_activities do
	feed = ActivityRepresenter.to_simple_form(activity, user, true)

	if feed do
	feed =
	ActivityRepresenter.wrap_with_entry(feed)
	\|> :xmerl.export_simple(:xmerl_xml)
	\|> to_string

	{:ok, private, _} = Keys.keys_from_pem(keys)
	{:ok, feed} = encode(private, feed)

	remote_users = remote_users(user, activity)

	salmon_urls = Enum.map(remote_users, & &1.info.salmon)
	reachable_urls_metadata = Instances.filter_reachable(salmon_urls)
	reachable_urls = Map.keys(reachable_urls_metadata)

	remote_users
	\|> Enum.filter(&(&1.info.salmon in reachable_urls))
	\|> Enum.each(fn remote_user ->
	Logger.debug(fn -> "Sending Salmon to #{remote_user.ap_id}" end)
	+ salmon_url = remote_user.info.salmon

	Publisher.enqueue_one(__MODULE__, %{
	recipient: remote_user,
	feed: feed,
	- unreachable_since: reachable_urls_metadata[remote_user.info.salmon]
	+ unreachable_since: reachable_urls_metadata[salmon_url],
	+ job: %{
	+ activity_id: activity.id,
	+ recipient: salmon_url
	+ }
	})
	end)
	end
	end

	def publish(%{id: id}, _), do: Logger.debug(fn -> "Keys missing for user #{id}" end)

	def gather_webfinger_links(%User{} = user) do
	{:ok, _private, public} = Keys.keys_from_pem(user.info.keys)
	magic_key = encode_key(public)

	[
	%{"rel" => "salmon", "href" => OStatus.salmon_path(user)},
	%{
	"rel" => "magic-public-key",
	"href" => "data:application/magic-public-key,#{magic_key}"
	}
	]
	end

	def gather_nodeinfo_protocol_names, do: []
	end
	diff --git a/lib/pleroma/web/websub/websub.ex b/lib/pleroma/web/websub/websub.ex
	index b61f388b8..57955ba4c 100644
	--- a/lib/pleroma/web/websub/websub.ex
	+++ b/lib/pleroma/web/websub/websub.ex
	@@ -1,332 +1,336 @@
	# 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.Activity
	alias Pleroma.HTTP
	alias Pleroma.Instances
	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.Visibility
	alias Pleroma.Web.Endpoint
	alias Pleroma.Web.Federator
	alias Pleroma.Web.Federator.Publisher
	alias Pleroma.Web.OStatus
	alias Pleroma.Web.OStatus.FeedRepresenter
	alias Pleroma.Web.Router.Helpers
	alias Pleroma.Web.Websub.WebsubClientSubscription
	alias Pleroma.Web.Websub.WebsubServerSubscription
	alias Pleroma.Web.XML
	require Logger

	import Ecto.Query

	@behaviour Pleroma.Web.Federator.Publisher

	def verify(subscription, getter \\ &HTTP.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 is_representable?(%Activity{data: %{"type" => type}} = activity)
	when type in @supported_activities,
	do: Visibility.is_public?(activity)

	def is_representable?(_), do: false

	def publish(topic, user, %{data: %{"type" => type}} = activity)
	when type in @supported_activities do
	response =
	user
	\|> FeedRepresenter.to_simple_form([activity], [user])
	\|> :xmerl.export_simple(:xmerl_xml)
	\|> to_string

	query =
	from(
	sub in WebsubServerSubscription,
	where: sub.topic == ^topic and sub.state == "active",
	where: fragment("? > (NOW() at time zone 'UTC')", sub.valid_until)
	)

	subscriptions = Repo.all(query)

	callbacks = Enum.map(subscriptions, & &1.callback)
	reachable_callbacks_metadata = Instances.filter_reachable(callbacks)
	reachable_callbacks = Map.keys(reachable_callbacks_metadata)

	subscriptions
	\|> Enum.filter(&(&1.callback in reachable_callbacks))
	\|> Enum.each(fn sub ->
	data = %{
	xml: response,
	topic: topic,
	callback: sub.callback,
	secret: sub.secret,
	- unreachable_since: reachable_callbacks_metadata[sub.callback]
	+ unreachable_since: reachable_callbacks_metadata[sub.callback],
	+ job: %{
	+ activity_id: activity.id,
	+ recipient: sub.callback
	+ }
	}

	Publisher.enqueue_one(__MODULE__, data)
	end)
	end

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

	def publish(actor, activity), do: publish(Pleroma.Web.OStatus.feed_path(actor), actor, activity)

	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)

	Federator.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 \\ &HTTP.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)
	preferred_username = XML.string_from_xpath("/feed/author[1]/poco:preferredUsername", doc)
	display_name = 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" => preferred_username \|\| name,
	"name" => display_name \|\| name,
	"host" => URI.parse(uri).host,
	"avatar" => avatar,
	"bio" => bio
	}}
	else
	e ->
	{:error, e}
	end
	end

	def request_subscription(websub, poster \\ &HTTP.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 ->
	Federator.request_subscription(sub)
	end)
	end

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

	with {:ok, %{status: code}} when code in 200..299 <-
	HTTP.post(
	callback,
	xml,
	[
	{"Content-Type", "application/atom+xml"},
	{"X-Hub-Signature", "sha1=#{signature}"}
	]
	) do
	if !Map.has_key?(params, :unreachable_since) \|\| params[:unreachable_since],
	do: Instances.set_reachable(callback)

	Logger.info(fn -> "Pushed to #{callback}, code #{code}" end)
	{:ok, code}
	else
	{_post_result, response} ->
	unless params[:unreachable_since], do: Instances.set_reachable(callback)
	Logger.debug(fn -> "Couldn't push to #{callback}, #{inspect(response)}" end)
	{:error, response}
	end
	end

	def gather_webfinger_links(%User{} = user) do
	[
	%{
	"rel" => "http://schemas.google.com/g/2010#updates-from",
	"type" => "application/atom+xml",
	"href" => OStatus.feed_path(user)
	},
	%{
	"rel" => "http://ostatus.org/schema/1.0/subscribe",
	"template" => OStatus.remote_follow_path()
	}
	]
	end

	def gather_nodeinfo_protocol_names, do: ["ostatus"]
	end
	diff --git a/priv/repo/migrations/20190719163442_create_federation_failures.exs b/priv/repo/migrations/20190719163442_create_federation_failures.exs
	new file mode 100644
	index 000000000..6143dd58a
	--- /dev/null
	+++ b/priv/repo/migrations/20190719163442_create_federation_failures.exs
	@@ -0,0 +1,17 @@
	+defmodule Pleroma.Repo.Migrations.CreateFederationFailures do
	+ use Ecto.Migration
	+
	+ def change do
	+ create_if_not_exists table(:federation_failures) do
	+ add(:activity_id, references(:activities, type: :uuid, on_delete: :delete_all))
	+ add(:recipient, :string, null: false)
	+ add(:transport, :string, null: false)
	+ add(:data, :map)
	+ add(:retries_count, :integer, default: 0)
	+
	+ timestamps()
	+ end
	+
	+ create_if_not_exists(unique_index(:federation_failures, [:activity_id, :recipient, :transport]))
	+ end
	+end

File Metadata

Mime Type: text/x-diff
Expires: Tue, Oct 13, 5:08 PM (1 d, 19 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1786833
Default Alt Text: (53 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions