Merge branch 'feature/federator-behaviours' into 'develop'

federator modularization (outgoing)

See merge request pleroma/pleroma!1140

config/config.exs

@@ -212,6 +212,11 @@
   registrations_open: true,
   federating: true,
   federation_reachability_timeout_days: 7,
+  federation_publisher_modules: [
+    Pleroma.Web.ActivityPub.Publisher,
+    Pleroma.Web.Websub,
+    Pleroma.Web.Salmon
+  ],
   allow_relay: true,
   rewrite_policy: Pleroma.Web.ActivityPub.MRF.NoOpPolicy,
   public: true,

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -5,7 +5,6 @@
 defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.Activity
   alias Pleroma.Conversation
-  alias Pleroma.Instances
   alias Pleroma.Notification
   alias Pleroma.Object
   alias Pleroma.Object.Fetcher
@@ -15,7 +14,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.MRF
   alias Pleroma.Web.ActivityPub.Transmogrifier
-  alias Pleroma.Web.Federator
   alias Pleroma.Web.WebFinger
 
   import Ecto.Query
@@ -24,8 +22,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
   require Logger
 
-  @httpoison Application.get_env(:pleroma, :httpoison)
-
   # For Announce activities, we filter the recipients based on following status for any actors
   # that match actual users.  See issue #164 for more information about why this is necessary.
   defp get_recipients(%{"type" => "Announce"} = data) do
@@ -961,89 +957,6 @@ def make_user_from_nickname(nickname) do
     end
   end
 
-  def should_federate?(inbox, public) do
-    if public do
-      true
-    else
-      inbox_info = URI.parse(inbox)
-      !Enum.member?(Pleroma.Config.get([:instance, :quarantined_instances], []), inbox_info.host)
-    end
-  end
-
-  def publish(actor, activity) do
-    remote_followers =
-      if actor.follower_address in activity.recipients do
-        {:ok, followers} = User.get_followers(actor)
-        followers |> Enum.filter(&(!&1.local))
-      else
-        []
-      end
-
-    public = is_public?(activity)
-
-    {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
-    json = Jason.encode!(data)
-
-    (Pleroma.Web.Salmon.remote_users(activity) ++ remote_followers)
-    |> 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} ->
-      Federator.publish_single_ap(%{
-        inbox: inbox,
-        json: json,
-        actor: actor,
-        id: activity.data["id"],
-        unreachable_since: unreachable_since
-      })
-    end)
-  end
-
-  def publish_one(%{inbox: inbox, json: json, actor: 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.Web.HTTPSignatures.sign(actor, %{
-        host: host,
-        "content-length": byte_size(json),
-        digest: digest,
-        date: date
-      })
-
-    with {:ok, %{status: code}} when code in 200..299 <-
-           result =
-             @httpoison.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
-
   # filter out broken threads
   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
     entire_thread_visible_for_user?(activity, user)

lib/pleroma/web/activity_pub/publisher.ex

@@ -0,0 +1,152 @@
+# 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.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
+
+  @httpoison Application.get_env(:pleroma, :httpoison)
+
+  @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.Web.HTTPSignatures.sign(actor, %{
+        host: host,
+        "content-length": byte_size(json),
+        digest: digest,
+        date: date
+      })
+
+    with {:ok, %{status: code}} when code in 200..299 <-
+           result =
+             @httpoison.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?(Pleroma.Config.get([:instance, :quarantined_instances], []), inbox_info.host)
+    end
+  end
+
+  @doc """
+  Publishes an activity to all relevant peers.
+  """
+  def publish(%User{} = actor, %Activity{} = activity) do
+    remote_followers =
+      if actor.follower_address in activity.recipients do
+        {:ok, followers} = User.get_followers(actor)
+        followers |> Enum.filter(&(!&1.local))
+      else
+        []
+      end
+
+    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)
+
+    (Pleroma.Web.Salmon.remote_users(activity) ++ remote_followers)
+    |> 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
+        }
+      )
+    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

lib/pleroma/web/federator/federator.ex

@@ -7,13 +7,10 @@ defmodule Pleroma.Web.Federator do
   alias Pleroma.Object.Containment
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ActivityPub
-  alias Pleroma.Web.ActivityPub.Relay
   alias Pleroma.Web.ActivityPub.Transmogrifier
   alias Pleroma.Web.ActivityPub.Utils
-  alias Pleroma.Web.ActivityPub.Visibility
+  alias Pleroma.Web.Federator.Publisher
   alias Pleroma.Web.Federator.RetryQueue
-  alias Pleroma.Web.OStatus
-  alias Pleroma.Web.Salmon
   alias Pleroma.Web.WebFinger
   alias Pleroma.Web.Websub
 
@@ -42,14 +39,6 @@ def publish(activity, priority \\ 1) do
     PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish, activity], priority)
   end
 
-  def publish_single_ap(params) do
-    PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_single_ap, params])
-  end
-
-  def publish_single_websub(websub) do
-    PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_single_websub, websub])
-  end
-
   def verify_websub(websub) do
     PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:verify_websub, websub])
   end
@@ -62,10 +51,6 @@ def refresh_subscriptions do
     PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:refresh_subscriptions])
   end
 
-  def publish_single_salmon(params) do
-    PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_single_salmon, params])
-  end
-
   # Job Worker Callbacks
 
   def perform(:refresh_subscriptions) do
@@ -95,23 +80,7 @@ def perform(:publish, activity) do
     with actor when not is_nil(actor) <- User.get_cached_by_ap_id(activity.data["actor"]) do
       {:ok, actor} = WebFinger.ensure_keys_present(actor)
 
-      if Visibility.is_public?(activity) do
+      Publisher.publish(actor, activity)
-        if OStatus.is_representable?(activity) do
-          Logger.info(fn -> "Sending #{activity.data["id"]} out via WebSub" end)
-          Websub.publish(Pleroma.Web.OStatus.feed_path(actor), actor, activity)
-
-          Logger.info(fn -> "Sending #{activity.data["id"]} out via Salmon" end)
-          Pleroma.Web.Salmon.publish(actor, activity)
-        end
-
-        if Keyword.get(Application.get_env(:pleroma, :instance), :allow_relay) do
-          Logger.info(fn -> "Relaying #{activity.data["id"]} out" end)
-          Relay.publish(activity)
-        end
-      end
-
-      Logger.info(fn -> "Sending #{activity.data["id"]} out via AP" end)
-      Pleroma.Web.ActivityPub.ActivityPub.publish(actor, activity)
     end
   end
 
@@ -153,20 +122,6 @@ def perform(:incoming_ap_doc, params) do
     end
   end
 
-  def perform(:publish_single_salmon, params) do
-    Salmon.send_to_user(params)
-  end
-
-  def perform(:publish_single_ap, params) do
-    case ActivityPub.publish_one(params) do
-      {:ok, _} ->
-        :ok
-
-      {:error, _} ->
-        RetryQueue.enqueue(params, ActivityPub)
-    end
-  end
-
   def perform(
         :publish_single_websub,
         %{xml: _xml, topic: _topic, callback: _callback, secret: _secret} = params

lib/pleroma/web/federator/publisher.ex

@@ -0,0 +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.Federator.Publisher do
+  alias Pleroma.Activity
+  alias Pleroma.Config
+  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(:federation_outgoing, __MODULE__, [:publish_one, module, params])
+
+  @spec perform(atom(), module(), any()) :: {:ok, any()} | {:error, any()}
+  def perform(:publish_one, module, params) do
+    case apply(module, :publish_one, [params]) do
+      {:ok, _} ->
+        :ok
+
+      {:error, _e} ->
+        RetryQueue.enqueue(params, module)
+    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(Pleroma.User.t(), Pleroma.Activity.t()) :: :ok | {:error, any()}
+
+  @spec publish(Pleroma.User.t(), Pleroma.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(Pleroma.User.t()) :: list()
+
+  @spec gather_webfinger_links(Pleroma.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

lib/pleroma/web/nodeinfo/nodeinfo_controller.ex

@@ -10,6 +10,7 @@ defmodule Pleroma.Web.Nodeinfo.NodeinfoController do
   alias Pleroma.User
   alias Pleroma.Web
   alias Pleroma.Web.ActivityPub.MRF
+  alias Pleroma.Web.Federator.Publisher
 
   plug(Pleroma.Web.FederatingPlug)
 
@@ -137,7 +138,7 @@ def raw_nodeinfo do
         name: Pleroma.Application.name() |> String.downcase(),
         version: Pleroma.Application.version()
       },
-      protocols: ["ostatus", "activitypub"],
+      protocols: Publisher.gather_nodeinfo_protocol_names(),
       services: %{
         inbound: [],
         outbound: []

lib/pleroma/web/ostatus/ostatus.ex

@@ -16,6 +16,7 @@ defmodule Pleroma.Web.OStatus do
   alias Pleroma.Web
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.ActivityPub.Transmogrifier
+  alias Pleroma.Web.ActivityPub.Visibility
   alias Pleroma.Web.OStatus.DeleteHandler
   alias Pleroma.Web.OStatus.FollowHandler
   alias Pleroma.Web.OStatus.NoteHandler
@@ -30,7 +31,7 @@ def is_representable?(%Activity{} = activity) do
       is_nil(object) ->
         false
 
-      object.data["type"] == "Note" ->
+      Visibility.is_public?(activity) && object.data["type"] == "Note" ->
         true
 
       true ->

lib/pleroma/web/salmon/salmon.ex

@@ -3,12 +3,18 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.Salmon do
+  @behaviour Pleroma.Web.Federator.Publisher
+
   @httpoison Application.get_env(:pleroma, :httpoison)
 
   use Bitwise
 
+  alias Pleroma.Activity
   alias Pleroma.Instances
   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
 
@@ -165,12 +171,12 @@ def remote_users(%{data: %{"to" => to} = data}) do
   end
 
   @doc "Pushes an activity to remote account."
-  def send_to_user(%{recipient: %{info: %{salmon: salmon}}} = params),
+  def publish_one(%{recipient: %{info: %{salmon: salmon}}} = params),
-    do: send_to_user(Map.put(params, :recipient, salmon))
+    do: publish_one(Map.put(params, :recipient, salmon))
 
-  def send_to_user(%{recipient: url, feed: feed, poster: poster} = params) when is_binary(url) do
+  def publish_one(%{recipient: url, feed: feed} = params) when is_binary(url) do
     with {:ok, %{status: code}} when code in 200..299 <-
-           poster.(
+           @httpoison.post(
              url,
              feed,
              [{"Content-Type", "application/magic-envelope+xml"}]
@@ -184,11 +190,11 @@ def send_to_user(%{recipient: url, feed: feed, poster: poster} = params) when is
       e ->
         unless params[:unreachable_since], do: Instances.set_reachable(url)
         Logger.debug(fn -> "Pushing Salmon to #{url} failed, #{inspect(e)}" end)
-        :error
+        {:error, "Unreachable instance"}
     end
   end
 
-  def send_to_user(_), do: :noop
+  def publish_one(_), do: :noop
 
   @supported_activities [
     "Create",
@@ -199,13 +205,19 @@ def send_to_user(_), do: :noop
     "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(), Pleroma.HTTP.t()) :: none
+  @spec publish(User.t(), Pleroma.Activity.t()) :: none
-  def publish(user, activity, poster \\ &@httpoison.post/3)
+  def publish(user, activity)
 
-  def publish(%{info: %{keys: keys}} = user, %{data: %{"type" => type}} = activity, poster)
+  def publish(%{info: %{keys: keys}} = user, %{data: %{"type" => type}} = activity)
       when type in @supported_activities do
     feed = ActivityRepresenter.to_simple_form(activity, user, true)
 
@@ -229,15 +241,29 @@ def publish(%{info: %{keys: keys}} = user, %{data: %{"type" => type}} = activity
       |> Enum.each(fn remote_user ->
         Logger.debug(fn -> "Sending Salmon to #{remote_user.ap_id}" end)
 
-        Pleroma.Web.Federator.publish_single_salmon(%{
+        Publisher.enqueue_one(__MODULE__, %{
           recipient: remote_user,
           feed: feed,
-          poster: poster,
           unreachable_since: reachable_urls_metadata[remote_user.info.salmon]
         })
       end)
     end
   end
 
-  def publish(%{id: id}, _, _), do: Logger.debug(fn -> "Keys missing for user #{id}" 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_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

lib/pleroma/web/web_finger/web_finger.ex

@@ -7,7 +7,7 @@ defmodule Pleroma.Web.WebFinger do
 
   alias Pleroma.User
   alias Pleroma.Web
-  alias Pleroma.Web.OStatus
+  alias Pleroma.Web.Federator.Publisher
   alias Pleroma.Web.Salmon
   alias Pleroma.Web.XML
   alias Pleroma.XmlBuilder
@@ -50,70 +50,40 @@ def webfinger(resource, fmt) when fmt in ["XML", "JSON"] do
     end
   end
 
+  defp gather_links(%User{} = user) do
+    [
+      %{
+        "rel" => "http://webfinger.net/rel/profile-page",
+        "type" => "text/html",
+        "href" => user.ap_id
+      }
+    ] ++ Publisher.gather_webfinger_links(user)
+  end
+
   def represent_user(user, "JSON") do
     {:ok, user} = ensure_keys_present(user)
-    {:ok, _private, public} = Salmon.keys_from_pem(user.info.keys)
-    magic_key = Salmon.encode_key(public)
 
     %{
       "subject" => "acct:#{user.nickname}@#{Pleroma.Web.Endpoint.host()}",
       "aliases" => [user.ap_id],
-      "links" => [
+      "links" => gather_links(user)
-        %{
-          "rel" => "http://schemas.google.com/g/2010#updates-from",
-          "type" => "application/atom+xml",
-          "href" => OStatus.feed_path(user)
-        },
-        %{
-          "rel" => "http://webfinger.net/rel/profile-page",
-          "type" => "text/html",
-          "href" => user.ap_id
-        },
-        %{"rel" => "salmon", "href" => OStatus.salmon_path(user)},
-        %{
-          "rel" => "magic-public-key",
-          "href" => "data:application/magic-public-key,#{magic_key}"
-        },
-        %{"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
-        },
-        %{
-          "rel" => "http://ostatus.org/schema/1.0/subscribe",
-          "template" => OStatus.remote_follow_path()
-        }
-      ]
     }
   end
 
   def represent_user(user, "XML") do
     {:ok, user} = ensure_keys_present(user)
-    {:ok, _private, public} = Salmon.keys_from_pem(user.info.keys)
+
-    magic_key = Salmon.encode_key(public)
+    links =
+      gather_links(user)
+      |> Enum.map(fn link -> {:Link, link} end)
 
     {
       :XRD,
       %{xmlns: "http://docs.oasis-open.org/ns/xri/xrd-1.0"},
       [
         {:Subject, "acct:#{user.nickname}@#{Pleroma.Web.Endpoint.host()}"},
-        {:Alias, user.ap_id},
+        {:Alias, user.ap_id}
-        {:Link,
+      ] ++ links
-         %{
-           rel: "http://schemas.google.com/g/2010#updates-from",
-           type: "application/atom+xml",
-           href: OStatus.feed_path(user)
-         }},
-        {:Link,
-         %{rel: "http://webfinger.net/rel/profile-page", type: "text/html", href: user.ap_id}},
-        {:Link, %{rel: "salmon", href: OStatus.salmon_path(user)}},
-        {:Link,
-         %{rel: "magic-public-key", href: "data:application/magic-public-key,#{magic_key}"}},
-        {:Link, %{rel: "self", type: "application/activity+json", href: user.ap_id}},
-        {:Link,
-         %{rel: "http://ostatus.org/schema/1.0/subscribe", template: OStatus.remote_follow_path()}}
-      ]
     }
     |> XmlBuilder.to_doc()
   end

lib/pleroma/web/websub/websub.ex

@@ -4,10 +4,14 @@
 
 defmodule Pleroma.Web.Websub do
   alias Ecto.Changeset
+  alias Pleroma.Activity
   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
@@ -18,6 +22,8 @@ defmodule Pleroma.Web.Websub do
 
   import Ecto.Query
 
+  @behaviour Pleroma.Web.Federator.Publisher
+
   @httpoison Application.get_env(:pleroma, :httpoison)
 
   def verify(subscription, getter \\ &@httpoison.get/3) do
@@ -56,6 +62,13 @@ def verify(subscription, getter \\ &@httpoison.get/3) do
     "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 =
@@ -88,12 +101,14 @@ def publish(topic, user, %{data: %{"type" => type}} = activity)
         unreachable_since: reachable_callbacks_metadata[sub.callback]
       }
 
-      Federator.publish_single_websub(data)
+      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
@@ -299,4 +314,20 @@ def publish_one(%{xml: xml, topic: topic, callback: callback, secret: secret} =
         {: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

lib/xml_builder.ex

@@ -35,6 +35,7 @@ def to_doc(content), do: ~s(<?xml version="1.0" encoding="UTF-8"?>) <> to_xml(co
   defp make_open_tag(tag, attributes) do
     attributes_string =
       for {attribute, value} <- attributes do
+        value = String.replace(value, "\"", "&quot;")
         "#{attribute}=\"#{value}\""
       end
       |> Enum.join(" ")

test/web/activity_pub/activity_pub_test.exs

@@ -10,6 +10,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
   alias Pleroma.Object
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ActivityPub
+  alias Pleroma.Web.ActivityPub.Publisher
   alias Pleroma.Web.ActivityPub.Utils
   alias Pleroma.Web.CommonAPI
 
@@ -963,8 +964,7 @@ test "it filters broken threads" do
 
       private_activity_1 = Activity.get_by_ap_id_with_object(private_activity_1.data["id"])
 
-      assert [public_activity, private_activity_1, private_activity_3] ==
+      assert [public_activity, private_activity_1, private_activity_3] == activities
-               activities
 
       assert length(activities) == 3
 
@@ -1057,7 +1057,7 @@ test "it can create a Flag activity" do
       actor = insert(:user)
       inbox = "http://200.site/users/nick1/inbox"
 
-      assert {:ok, _} = ActivityPub.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1})
+      assert {:ok, _} = Publisher.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1})
 
       assert called(Instances.set_reachable(inbox))
     end
@@ -1070,7 +1070,7 @@ test "it can create a Flag activity" do
       inbox = "http://200.site/users/nick1/inbox"
 
       assert {:ok, _} =
-               ActivityPub.publish_one(%{
+               Publisher.publish_one(%{
                  inbox: inbox,
                  json: "{}",
                  actor: actor,
@@ -1089,7 +1089,7 @@ test "it can create a Flag activity" do
       inbox = "http://200.site/users/nick1/inbox"
 
       assert {:ok, _} =
-               ActivityPub.publish_one(%{
+               Publisher.publish_one(%{
                  inbox: inbox,
                  json: "{}",
                  actor: actor,
@@ -1107,8 +1107,7 @@ test "it can create a Flag activity" do
       actor = insert(:user)
       inbox = "http://404.site/users/nick1/inbox"
 
-      assert {:error, _} =
+      assert {:error, _} = Publisher.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1})
-               ActivityPub.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1})
 
       assert called(Instances.set_unreachable(inbox))
     end
@@ -1120,8 +1119,7 @@ test "it can create a Flag activity" do
       actor = insert(:user)
       inbox = "http://connrefused.site/users/nick1/inbox"
 
-      assert {:error, _} =
+      assert {:error, _} = Publisher.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1})
-               ActivityPub.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1})
 
       assert called(Instances.set_unreachable(inbox))
     end
@@ -1133,7 +1131,7 @@ test "it can create a Flag activity" do
       actor = insert(:user)
       inbox = "http://200.site/users/nick1/inbox"
 
-      assert {:ok, _} = ActivityPub.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1})
+      assert {:ok, _} = Publisher.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1})
 
       refute called(Instances.set_unreachable(inbox))
     end
@@ -1146,7 +1144,7 @@ test "it can create a Flag activity" do
       inbox = "http://connrefused.site/users/nick1/inbox"
 
       assert {:error, _} =
-               ActivityPub.publish_one(%{
+               Publisher.publish_one(%{
                  inbox: inbox,
                  json: "{}",
                  actor: actor,

test/web/federator_test.exs

@@ -58,7 +58,7 @@ test "with relays deactivated, it does not publish to the relay", %{
 
   describe "Targets reachability filtering in `publish`" do
     test_with_mock "it federates only to reachable instances via AP",
-                   Federator,
+                   Pleroma.Web.ActivityPub.Publisher,
                    [:passthrough],
                    [] do
       user = insert(:user)
@@ -88,13 +88,18 @@ test "with relays deactivated, it does not publish to the relay", %{
       {:ok, _activity} =
         CommonAPI.post(user, %{"status" => "HI @nick1@domain.com, @nick2@domain2.com!"})
 
-      assert called(Federator.publish_single_ap(%{inbox: inbox1, unreachable_since: dt}))
+      assert called(
+               Pleroma.Web.ActivityPub.Publisher.publish_one(%{
+                 inbox: inbox1,
+                 unreachable_since: dt
+               })
+             )
 
-      refute called(Federator.publish_single_ap(%{inbox: inbox2}))
+      refute called(Pleroma.Web.ActivityPub.Publisher.publish_one(%{inbox: inbox2}))
     end
 
     test_with_mock "it federates only to reachable instances via Websub",
-                   Federator,
+                   Pleroma.Web.Websub,
                    [:passthrough],
                    [] do
       user = insert(:user)
@@ -122,17 +127,17 @@ test "with relays deactivated, it does not publish to the relay", %{
       {:ok, _activity} = CommonAPI.post(user, %{"status" => "HI"})
 
       assert called(
-               Federator.publish_single_websub(%{
+               Pleroma.Web.Websub.publish_one(%{
                  callback: sub2.callback,
                  unreachable_since: dt
                })
              )
 
-      refute called(Federator.publish_single_websub(%{callback: sub1.callback}))
+      refute called(Pleroma.Web.Websub.publish_one(%{callback: sub1.callback}))
     end
 
     test_with_mock "it federates only to reachable instances via Salmon",
-                   Federator,
+                   Pleroma.Web.Salmon,
                    [:passthrough],
                    [] do
       user = insert(:user)
@@ -162,13 +167,13 @@ test "with relays deactivated, it does not publish to the relay", %{
         CommonAPI.post(user, %{"status" => "HI @nick1@domain.com, @nick2@domain2.com!"})
 
       assert called(
-               Federator.publish_single_salmon(%{
+               Pleroma.Web.Salmon.publish_one(%{
                  recipient: remote_user2,
                  unreachable_since: dt
                })
              )
 
-      refute called(Federator.publish_single_websub(%{recipient: remote_user1}))
+      refute called(Pleroma.Web.Salmon.publish_one(%{recipient: remote_user1}))
     end
   end
 

test/web/salmon/salmon_test.exs

@@ -7,7 +7,9 @@ defmodule Pleroma.Web.Salmon.SalmonTest do
   alias Pleroma.Activity
   alias Pleroma.Repo
   alias Pleroma.User
+  alias Pleroma.Web.Federator.Publisher
   alias Pleroma.Web.Salmon
+  import Mock
   import Pleroma.Factory
 
   @magickey "RSA.pu0s-halox4tu7wmES1FVSx6u-4wc0YrUFXcqWXZG4-27UmbCOpMQftRCldNRfyA-qLbz-eqiwQhh-1EwUvjsD4cYbAHNGHwTvDOyx5AKthQUP44ykPv7kjKGh3DWKySJvcs9tlUG87hlo7AvnMo9pwRS_Zz2CacQ-MKaXyDepk=.AQAB"
@@ -77,7 +79,10 @@ test "it gets a magic key" do
              "RSA.uzg6r1peZU0vXGADWxGJ0PE34WvmhjUmydbX5YYdOiXfODVLwCMi1umGoqUDm-mRu4vNEdFBVJU1CpFA7dKzWgIsqsa501i2XqElmEveXRLvNRWFB6nG03Q5OUY2as8eE54BJm0p20GkMfIJGwP6TSFb-ICp3QjzbatuSPJ6xCE=.AQAB"
   end
 
-  test "it pushes an activity to remote accounts it's addressed to" do
+  test_with_mock "it pushes an activity to remote accounts it's addressed to",
+                 Publisher,
+                 [:passthrough],
+                 [] do
     user_data = %{
       info: %{
         salmon: "http://test-example.org/salmon"
@@ -102,10 +107,8 @@ test "it pushes an activity to remote accounts it's addressed to" do
     user = User.get_cached_by_ap_id(activity.data["actor"])
     {:ok, user} = Pleroma.Web.WebFinger.ensure_keys_present(user)
 
-    poster = fn url, _data, _headers ->
+    Salmon.publish(user, activity)
-      assert url == "http://test-example.org/salmon"
-    end
 
-    Salmon.publish(user, activity, poster)
+    assert called(Publisher.enqueue_one(Salmon, %{recipient: mentioned_user}))
   end
 end