Merge branch '534_federation_targets_reachability' into 'develop'

[#534] Unreachable federation targets retirement

Closes #534

See merge request pleroma/pleroma!703

config/config.exs

@@ -146,6 +146,7 @@
   banner_upload_limit: 4_000_000,
   registrations_open: true,
   federating: true,
+  federation_reachability_timeout_days: 7,
   allow_relay: true,
   rewrite_policy: Pleroma.Web.ActivityPub.MRF.NoOpPolicy,
   public: true,

docs/config.md

@@ -72,6 +72,7 @@ config :pleroma, Pleroma.Mailer,
 * `invites_enabled`: Enable user invitations for admins (depends on `registrations_open: false`).
 * `account_activation_required`: Require users to confirm their emails before signing in.
 * `federating`: Enable federation with other instances
+* `federation_reachability_timeout_days`: Timeout (in days) of each external federation target being unreachable prior to pausing federating to it.
 * `allow_relay`: Enable Pleroma’s Relay, which makes it possible to follow a whole instance
 * `rewrite_policy`: Message Rewrite Policy, either one or a list. Here are the ones available by default:
   * `Pleroma.Web.ActivityPub.MRF.NoOpPolicy`: Doesn’t modify activities (default)

lib/pleroma/instances.ex

@@ -0,0 +1,36 @@
+defmodule Pleroma.Instances do
+  @moduledoc "Instances context."
+
+  @adapter Pleroma.Instances.Instance
+
+  defdelegate filter_reachable(urls_or_hosts), to: @adapter
+  defdelegate reachable?(url_or_host), to: @adapter
+  defdelegate set_reachable(url_or_host), to: @adapter
+  defdelegate set_unreachable(url_or_host, unreachable_since \\ nil), to: @adapter
+
+  def set_consistently_unreachable(url_or_host),
+    do: set_unreachable(url_or_host, reachability_datetime_threshold())
+
+  def reachability_datetime_threshold do
+    federation_reachability_timeout_days =
+      Pleroma.Config.get(:instance)[:federation_reachability_timeout_days] || 0
+
+    if federation_reachability_timeout_days > 0 do
+      NaiveDateTime.add(
+        NaiveDateTime.utc_now(),
+        -federation_reachability_timeout_days * 24 * 3600,
+        :second
+      )
+    else
+      ~N[0000-01-01 00:00:00]
+    end
+  end
+
+  def host(url_or_host) when is_binary(url_or_host) do
+    if url_or_host =~ ~r/^http/i do
+      URI.parse(url_or_host).host
+    else
+      url_or_host
+    end
+  end
+end

lib/pleroma/instances/instance.ex

@@ -0,0 +1,102 @@
+defmodule Pleroma.Instances.Instance do
+  @moduledoc "Instance."
+
+  alias Pleroma.Instances
+  alias Pleroma.Instances.Instance
+
+  use Ecto.Schema
+
+  import Ecto.{Query, Changeset}
+
+  alias Pleroma.Repo
+
+  schema "instances" do
+    field(:host, :string)
+    field(:unreachable_since, :naive_datetime)
+
+    timestamps()
+  end
+
+  defdelegate host(url_or_host), to: Instances
+
+  def changeset(struct, params \\ %{}) do
+    struct
+    |> cast(params, [:host, :unreachable_since])
+    |> validate_required([:host])
+    |> unique_constraint(:host)
+  end
+
+  def filter_reachable([]), do: []
+
+  def filter_reachable(urls_or_hosts) when is_list(urls_or_hosts) do
+    hosts =
+      urls_or_hosts
+      |> Enum.map(&(&1 && host(&1)))
+      |> Enum.filter(&(to_string(&1) != ""))
+
+    unreachable_hosts =
+      Repo.all(
+        from(i in Instance,
+          where:
+            i.host in ^hosts and
+              i.unreachable_since <= ^Instances.reachability_datetime_threshold(),
+          select: i.host
+        )
+      )
+
+    Enum.filter(urls_or_hosts, &(&1 && host(&1) not in unreachable_hosts))
+  end
+
+  def reachable?(url_or_host) when is_binary(url_or_host) do
+    !Repo.one(
+      from(i in Instance,
+        where:
+          i.host == ^host(url_or_host) and
+            i.unreachable_since <= ^Instances.reachability_datetime_threshold(),
+        select: true
+      )
+    )
+  end
+
+  def reachable?(_), do: true
+
+  def set_reachable(url_or_host) when is_binary(url_or_host) do
+    with host <- host(url_or_host),
+         %Instance{} = existing_record <- Repo.get_by(Instance, %{host: host}) do
+      {:ok, _instance} =
+        existing_record
+        |> changeset(%{unreachable_since: nil})
+        |> Repo.update()
+    end
+  end
+
+  def set_reachable(_), do: {:error, nil}
+
+  def set_unreachable(url_or_host, unreachable_since \\ nil)
+
+  def set_unreachable(url_or_host, unreachable_since) when is_binary(url_or_host) do
+    unreachable_since = unreachable_since || DateTime.utc_now()
+    host = host(url_or_host)
+    existing_record = Repo.get_by(Instance, %{host: host})
+
+    changes = %{unreachable_since: unreachable_since}
+
+    cond do
+      is_nil(existing_record) ->
+        %Instance{}
+        |> changeset(Map.put(changes, :host, host))
+        |> Repo.insert()
+
+      existing_record.unreachable_since &&
+          NaiveDateTime.compare(existing_record.unreachable_since, unreachable_since) != :gt ->
+        {:ok, existing_record}
+
+      true ->
+        existing_record
+        |> changeset(changes)
+        |> Repo.update()
+    end
+  end
+
+  def set_unreachable(_, _), do: {:error, nil}
+end

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -3,7 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.ActivityPub do
-  alias Pleroma.{Activity, Repo, Object, Upload, User, Notification}
+  alias Pleroma.{Activity, Repo, Object, Upload, User, Notification, Instances}
   alias Pleroma.Web.ActivityPub.{Transmogrifier, MRF}
   alias Pleroma.Web.WebFinger
   alias Pleroma.Web.Federator
@@ -734,7 +734,7 @@ def should_federate?(inbox, public) do
   end
 
   def publish(actor, activity) do
-    followers =
+    remote_followers =
       if actor.follower_address in activity.recipients do
         {:ok, followers} = User.get_followers(actor)
         followers |> Enum.filter(&(!&1.local))
@@ -745,13 +745,14 @@ def publish(actor, activity) do
     public = is_public?(activity)
 
     remote_inboxes =
-      (Pleroma.Web.Salmon.remote_users(activity) ++ followers)
+      (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()
 
     {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
     json = Jason.encode!(data)
@@ -779,15 +780,24 @@ def publish_one(%{inbox: inbox, json: json, actor: actor, id: id}) do
         digest: digest
       })
 
-    @httpoison.post(
+    with {:ok, %{status: code}} when code in 200..299 <-
-      inbox,
+           result =
-      json,
+             @httpoison.post(
-      [
+               inbox,
-        {"Content-Type", "application/activity+json"},
+               json,
-        {"signature", signature},
+               [
-        {"digest", digest}
+                 {"Content-Type", "application/activity+json"},
-      ]
+                 {"signature", signature},
-    )
+                 {"digest", digest}
+               ]
+             ) do
+      Instances.set_reachable(inbox)
+      result
+    else
+      {_post_result, response} ->
+        Instances.set_unreachable(inbox)
+        {:error, response}
+    end
   end
 
   # TODO:

lib/pleroma/web/activity_pub/activity_pub_controller.ex

@@ -4,6 +4,7 @@
 
 defmodule Pleroma.Web.ActivityPub.ActivityPubController do
   use Pleroma.Web, :controller
+
   alias Pleroma.{Activity, User, Object}
   alias Pleroma.Web.ActivityPub.{ObjectView, UserView}
   alias Pleroma.Web.ActivityPub.ActivityPub
@@ -17,6 +18,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
   action_fallback(:errors)
 
   plug(Pleroma.Web.FederatingPlug when action in [:inbox, :relay])
+  plug(:set_requester_reachable when action in [:inbox])
   plug(:relay_active? when action in [:relay])
 
   def relay_active?(conn, _) do
@@ -289,4 +291,13 @@ def errors(conn, _e) do
     |> put_status(500)
     |> json("error")
   end
+
+  defp set_requester_reachable(%Plug.Conn{} = conn, _) do
+    with actor <- conn.params["actor"],
+         true <- is_binary(actor) do
+      Pleroma.Instances.set_reachable(actor)
+    end
+
+    conn
+  end
 end

lib/pleroma/web/federator/federator.ex

@@ -6,7 +6,7 @@ defmodule Pleroma.Web.Federator do
   use GenServer
   alias Pleroma.User
   alias Pleroma.Activity
-  alias Pleroma.Web.{WebFinger, Websub}
+  alias Pleroma.Web.{WebFinger, Websub, Salmon}
   alias Pleroma.Web.Federator.RetryQueue
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.ActivityPub.Relay
@@ -124,6 +124,10 @@ def handle(:incoming_ap_doc, params) do
     end
   end
 
+  def handle(:publish_single_salmon, {user_or_url, feed, poster}) do
+    Salmon.send_to_user(user_or_url, feed, poster)
+  end
+
   def handle(:publish_single_ap, params) do
     case ActivityPub.publish_one(params) do
       {:ok, _} ->

lib/pleroma/web/ostatus/ostatus.ex

@@ -48,6 +48,9 @@ def remote_follow_path do
 
   def handle_incoming(xml_string) do
     with doc when doc != :error <- parse_document(xml_string) do
+      with {:ok, actor_user} <- find_make_or_update_user(doc),
+           do: Pleroma.Instances.set_reachable(actor_user.ap_id)
+
       entries = :xmerl_xpath.string('//entry', doc)
 
       activities =

lib/pleroma/web/ostatus/ostatus_controller.ex

@@ -14,6 +14,7 @@ defmodule Pleroma.Web.OStatus.OStatusController do
   alias Pleroma.Web.ActivityPub.ActivityPub
 
   plug(Pleroma.Web.FederatingPlug when action in [:salmon_incoming])
+
   action_fallback(:errors)
 
   def feed_redirect(conn, %{"nickname" => nickname}) do

lib/pleroma/web/salmon/salmon.ex

@@ -6,6 +6,7 @@ defmodule Pleroma.Web.Salmon do
   @httpoison Application.get_env(:pleroma, :httpoison)
 
   use Bitwise
+  alias Pleroma.Instances
   alias Pleroma.Web.XML
   alias Pleroma.Web.OStatus.ActivityRepresenter
   alias Pleroma.User
@@ -163,23 +164,28 @@ def remote_users(%{data: %{"to" => to} = data}) do
 
   # push an activity to remote accounts
   #
-  defp send_to_user(%{info: %{salmon: salmon}}, feed, poster),
+  def send_to_user(%{info: %{salmon: salmon}}, feed, poster),
     do: send_to_user(salmon, feed, poster)
 
-  defp send_to_user(url, feed, poster) when is_binary(url) do
+  def send_to_user(url, feed, poster) when is_binary(url) do
-    with {:ok, %{status: code}} <-
+    with {:ok, %{status: code}} when code in 200..299 <-
            poster.(
              url,
              feed,
              [{"Content-Type", "application/magic-envelope+xml"}]
            ) do
+      Instances.set_reachable(url)
       Logger.debug(fn -> "Pushed to #{url}, code #{code}" end)
+      :ok
     else
-      e -> Logger.debug(fn -> "Pushing Salmon to #{url} failed, #{inspect(e)}" end)
+      e ->
+        Instances.set_unreachable(url)
+        Logger.debug(fn -> "Pushing Salmon to #{url} failed, #{inspect(e)}" end)
+        :error
     end
   end
 
-  defp send_to_user(_, _, _), do: nil
+  def send_to_user(_, _, _), do: :noop
 
   @supported_activities [
     "Create",
@@ -209,12 +215,16 @@ def publish(%{info: %{keys: keys}} = user, %{data: %{"type" => type}} = activity
       {:ok, private, _} = keys_from_pem(keys)
       {:ok, feed} = encode(private, feed)
 
-      remote_users(activity)
+      remote_users = remote_users(activity)
+
+      salmon_urls = Enum.map(remote_users, & &1.info.salmon)
+      reachable_salmon_urls = Instances.filter_reachable(salmon_urls)
+
+      remote_users
+      |> Enum.filter(&(&1.info.salmon in reachable_salmon_urls))
       |> Enum.each(fn remote_user ->
-        Task.start(fn ->
+        Logger.debug(fn -> "Sending Salmon to #{remote_user.ap_id}" end)
-          Logger.debug(fn -> "Sending Salmon to #{remote_user.ap_id}" end)
+        Pleroma.Web.Federator.enqueue(:publish_single_salmon, {remote_user, feed, poster})
-          send_to_user(remote_user, feed, poster)
-        end)
       end)
     end
   end

lib/pleroma/web/websub/websub.ex

@@ -5,6 +5,7 @@
 defmodule Pleroma.Web.Websub do
   alias Ecto.Changeset
   alias Pleroma.Repo
+  alias Pleroma.Instances
   alias Pleroma.Web.Websub.{WebsubServerSubscription, WebsubClientSubscription}
   alias Pleroma.Web.OStatus.FeedRepresenter
   alias Pleroma.Web.{XML, Endpoint, OStatus}
@@ -53,23 +54,27 @@ def verify(subscription, getter \\ &@httpoison.get/3) do
   ]
   def publish(topic, user, %{data: %{"type" => type}} = activity)
       when type in @supported_activities do
-    # TODO: Only send to still valid subscriptions.
+    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()", sub.valid_until)
+        where: fragment("? > (NOW() at time zone 'UTC')", sub.valid_until)
       )
 
     subscriptions = Repo.all(query)
 
-    Enum.each(subscriptions, fn sub ->
+    callbacks = Enum.map(subscriptions, & &1.callback)
-      response =
+    reachable_callbacks = Instances.filter_reachable(callbacks)
-        user
-        |> FeedRepresenter.to_simple_form([activity], [user])
-        |> :xmerl.export_simple(:xmerl_xml)
-        |> to_string
 
+    subscriptions
+    |> Enum.filter(&(&1.callback in reachable_callbacks))
+    |> Enum.each(fn sub ->
       data = %{
         xml: response,
         topic: topic,
@@ -267,7 +272,7 @@ def publish_one(%{xml: xml, topic: topic, callback: callback, secret: secret}) d
     signature = sign(secret || "", xml)
     Logger.info(fn -> "Pushing #{topic} to #{callback}" end)
 
-    with {:ok, %{status: code}} <-
+    with {:ok, %{status: code}} when code in 200..299 <-
            @httpoison.post(
              callback,
              xml,
@@ -276,12 +281,14 @@ def publish_one(%{xml: xml, topic: topic, callback: callback, secret: secret}) d
                {"X-Hub-Signature", "sha1=#{signature}"}
              ]
            ) do
+      Instances.set_reachable(callback)
       Logger.info(fn -> "Pushed to #{callback}, code #{code}" end)
       {:ok, code}
     else
-      e ->
+      {_post_result, response} ->
-        Logger.debug(fn -> "Couldn't push to #{callback}, #{inspect(e)}" end)
+        Instances.set_unreachable(callback)
-        {:error, e}
+        Logger.debug(fn -> "Couldn't push to #{callback}, #{inspect(response)}" end)
+        {:error, response}
     end
   end
 end

lib/pleroma/web/websub/websub_controller.ex

@@ -4,9 +4,11 @@
 
 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(

priv/repo/migrations/20190123125546_create_instances.exs

@@ -0,0 +1,15 @@
+defmodule Pleroma.Repo.Migrations.CreateInstances do
+  use Ecto.Migration
+
+  def change do
+    create table(:instances) do
+      add :host, :string
+      add :unreachable_since, :naive_datetime
+
+      timestamps()
+    end
+
+    create unique_index(:instances, [:host])
+    create index(:instances, [:unreachable_since])
+  end
+end

test/support/factory.ex

@@ -193,7 +193,7 @@ def follow_activity_factory do
   def websub_subscription_factory do
     %Pleroma.Web.Websub.WebsubServerSubscription{
       topic: "http://example.org",
-      callback: "http://example/org/callback",
+      callback: "http://example.org/callback",
       secret: "here's a secret",
       valid_until: NaiveDateTime.add(NaiveDateTime.utc_now(), 100),
       state: "requested"
@@ -220,4 +220,11 @@ def oauth_app_factory do
       client_secret: "aaa;/&bbb"
     }
   end
+
+  def instance_factory do
+    %Pleroma.Instances.Instance{
+      host: "domain.com",
+      unreachable_since: nil
+    }
+  end
 end

test/support/http_request_mock.ex

@@ -696,6 +696,14 @@ def get("http://example.com/empty", _, _, _) do
     {:ok, %Tesla.Env{status: 200, body: "hello"}}
   end
 
+  def get("http://404.site" <> _, _, _, _) do
+    {:ok,
+     %Tesla.Env{
+       status: 404,
+       body: ""
+     }}
+  end
+
   def get(url, query, body, headers) do
     {:error,
      "Not implemented the mock response for get #{inspect(url)}, #{query}, #{inspect(body)}, #{
@@ -716,6 +724,26 @@ def post("http://example.org/needs_refresh", _, _, _) do
      }}
   end
 
+  def post("http://200.site" <> _, _, _, _) do
+    {:ok,
+     %Tesla.Env{
+       status: 200,
+       body: ""
+     }}
+  end
+
+  def post("http://connrefused.site" <> _, _, _, _) do
+    {:error, :connrefused}
+  end
+
+  def post("http://404.site" <> _, _, _, _) do
+    {:ok,
+     %Tesla.Env{
+       status: 404,
+       body: ""
+     }}
+  end
+
   def post(url, _query, _body, _headers) do
     {:error, "Not implemented the mock response for post #{inspect(url)}"}
   end

test/web/activity_pub/activity_pub_controller_test.exs

@@ -6,8 +6,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubControllerTest do
   use Pleroma.Web.ConnCase
   import Pleroma.Factory
   alias Pleroma.Web.ActivityPub.{UserView, ObjectView}
-  alias Pleroma.{Object, Repo, User}
+  alias Pleroma.{Object, Repo, Activity, User, Instances}
-  alias Pleroma.Activity
 
   setup_all do
     Tesla.Mock.mock_global(fn env -> apply(HttpRequestMock, :request, [env]) end)
@@ -144,6 +143,23 @@ test "it inserts an incoming activity into the database", %{conn: conn} do
       :timer.sleep(500)
       assert Activity.get_by_ap_id(data["id"])
     end
+
+    test "it clears `unreachable` federation status of the sender", %{conn: conn} do
+      data = File.read!("test/fixtures/mastodon-post-activity.json") |> Poison.decode!()
+
+      sender_url = data["actor"]
+      Instances.set_consistently_unreachable(sender_url)
+      refute Instances.reachable?(sender_url)
+
+      conn =
+        conn
+        |> assign(:valid_signature, true)
+        |> put_req_header("content-type", "application/activity+json")
+        |> post("/inbox", data)
+
+      assert "ok" == json_response(conn, 200)
+      assert Instances.reachable?(sender_url)
+    end
   end
 
   describe "/users/:nickname/inbox" do
@@ -191,6 +207,28 @@ test "it returns a note activity in a collection", %{conn: conn} do
 
       assert response(conn, 200) =~ note_activity.data["object"]["content"]
     end
+
+    test "it clears `unreachable` federation status of the sender", %{conn: conn} do
+      user = insert(:user)
+
+      data =
+        File.read!("test/fixtures/mastodon-post-activity.json")
+        |> Poison.decode!()
+        |> Map.put("bcc", [user.ap_id])
+
+      sender_host = URI.parse(data["actor"]).host
+      Instances.set_consistently_unreachable(sender_host)
+      refute Instances.reachable?(sender_host)
+
+      conn =
+        conn
+        |> assign(:valid_signature, true)
+        |> put_req_header("content-type", "application/activity+json")
+        |> post("/users/#{user.nickname}/inbox", data)
+
+      assert "ok" == json_response(conn, 200)
+      assert Instances.reachable?(sender_host)
+    end
   end
 
   describe "/users/:nickname/outbox" do

test/web/activity_pub/activity_pub_test.exs

@@ -7,11 +7,12 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.ActivityPub.Utils
   alias Pleroma.Web.CommonAPI
-  alias Pleroma.{Activity, Object, User}
+  alias Pleroma.{Activity, Object, User, Instances}
   alias Pleroma.Builders.ActivityBuilder
 
   import Pleroma.Factory
   import Tesla.Mock
+  import Mock
 
   setup do
     mock(fn env -> apply(HttpRequestMock, :request, [env]) end)
@@ -696,6 +697,46 @@ test "returned pinned statuses" do
     assert 3 = length(activities)
   end
 
+  describe "publish_one/1" do
+    test_with_mock "it calls `Instances.set_unreachable` on target inbox on non-2xx HTTP response code",
+                   Instances,
+                   [:passthrough],
+                   [] do
+      actor = insert(:user)
+      inbox = "http://404.site/users/nick1/inbox"
+
+      assert {:error, _} =
+               ActivityPub.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1})
+
+      assert called(Instances.set_unreachable(inbox))
+    end
+
+    test_with_mock "it calls `Instances.set_unreachable` on target inbox on request error of any kind",
+                   Instances,
+                   [:passthrough],
+                   [] do
+      actor = insert(:user)
+      inbox = "http://connrefused.site/users/nick1/inbox"
+
+      assert {:error, _} =
+               ActivityPub.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1})
+
+      assert called(Instances.set_unreachable(inbox))
+    end
+
+    test_with_mock "it does NOT call `Instances.set_unreachable` if target is reachable",
+                   Instances,
+                   [:passthrough],
+                   [] do
+      actor = insert(:user)
+      inbox = "http://200.site/users/nick1/inbox"
+
+      assert {:ok, _} = ActivityPub.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1})
+
+      refute called(Instances.set_unreachable(inbox))
+    end
+  end
+
   def data_uri do
     File.read!("test/fixtures/avatar_data_uri")
   end

test/web/federator_test.exs

@@ -3,8 +3,8 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.FederatorTest do
-  alias Pleroma.Web.Federator
+  alias Pleroma.Web.{CommonAPI, Federator}
-  alias Pleroma.Web.CommonAPI
+  alias Pleroma.Instances
   use Pleroma.DataCase
   import Pleroma.Factory
   import Mock
@@ -71,6 +71,103 @@ test "with relays deactivated, it does not publish to the relay", %{
     end
   end
 
+  describe "Targets reachability filtering in `publish`" do
+    test_with_mock "it federates only to reachable instances via AP",
+                   Federator,
+                   [:passthrough],
+                   [] do
+      user = insert(:user)
+
+      {inbox1, inbox2} =
+        {"https://domain.com/users/nick1/inbox", "https://domain2.com/users/nick2/inbox"}
+
+      insert(:user, %{
+        local: false,
+        nickname: "nick1@domain.com",
+        ap_id: "https://domain.com/users/nick1",
+        info: %{ap_enabled: true, source_data: %{"inbox" => inbox1}}
+      })
+
+      insert(:user, %{
+        local: false,
+        nickname: "nick2@domain2.com",
+        ap_id: "https://domain2.com/users/nick2",
+        info: %{ap_enabled: true, source_data: %{"inbox" => inbox2}}
+      })
+
+      Instances.set_unreachable(
+        URI.parse(inbox2).host,
+        Instances.reachability_datetime_threshold()
+      )
+
+      {:ok, _activity} =
+        CommonAPI.post(user, %{"status" => "HI @nick1@domain.com, @nick2@domain2.com!"})
+
+      assert called(Federator.enqueue(:publish_single_ap, %{inbox: inbox1}))
+      refute called(Federator.enqueue(:publish_single_ap, %{inbox: inbox2}))
+    end
+
+    test_with_mock "it federates only to reachable instances via Websub",
+                   Federator,
+                   [:passthrough],
+                   [] do
+      user = insert(:user)
+      websub_topic = Pleroma.Web.OStatus.feed_path(user)
+
+      sub1 =
+        insert(:websub_subscription, %{
+          topic: websub_topic,
+          state: "active",
+          callback: "http://pleroma.soykaf.com/cb"
+        })
+
+      sub2 =
+        insert(:websub_subscription, %{
+          topic: websub_topic,
+          state: "active",
+          callback: "https://pleroma2.soykaf.com/cb"
+        })
+
+      Instances.set_consistently_unreachable(sub1.callback)
+
+      {:ok, _activity} = CommonAPI.post(user, %{"status" => "HI"})
+
+      assert called(Federator.enqueue(:publish_single_websub, %{callback: sub2.callback}))
+      refute called(Federator.enqueue(:publish_single_websub, %{callback: sub1.callback}))
+    end
+
+    test_with_mock "it federates only to reachable instances via Salmon",
+                   Federator,
+                   [:passthrough],
+                   [] do
+      user = insert(:user)
+
+      remote_user1 =
+        insert(:user, %{
+          local: false,
+          nickname: "nick1@domain.com",
+          ap_id: "https://domain.com/users/nick1",
+          info: %{salmon: "https://domain.com/salmon"}
+        })
+
+      remote_user2 =
+        insert(:user, %{
+          local: false,
+          nickname: "nick2@domain2.com",
+          ap_id: "https://domain2.com/users/nick2",
+          info: %{salmon: "https://domain2.com/salmon"}
+        })
+
+      Instances.set_consistently_unreachable("domain.com")
+
+      {:ok, _activity} =
+        CommonAPI.post(user, %{"status" => "HI @nick1@domain.com, @nick2@domain2.com!"})
+
+      assert called(Federator.enqueue(:publish_single_salmon, {remote_user2, :_, :_}))
+      refute called(Federator.enqueue(:publish_single_websub, {remote_user1, :_, :_}))
+    end
+  end
+
   describe "Receive an activity" do
     test "successfully processes incoming AP docs with correct origin" do
       params = %{

test/web/instances/instance_test.exs

@@ -0,0 +1,107 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2018 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Instances.InstanceTest do
+  alias Pleroma.Repo
+  alias Pleroma.Instances.Instance
+
+  use Pleroma.DataCase
+
+  import Pleroma.Factory
+
+  setup_all do
+    config_path = [:instance, :federation_reachability_timeout_days]
+    initial_setting = Pleroma.Config.get(config_path)
+
+    Pleroma.Config.put(config_path, 1)
+    on_exit(fn -> Pleroma.Config.put(config_path, initial_setting) end)
+
+    :ok
+  end
+
+  describe "set_reachable/1" do
+    test "clears `unreachable_since` of existing matching Instance record having non-nil `unreachable_since`" do
+      instance = insert(:instance, unreachable_since: NaiveDateTime.utc_now())
+
+      assert {:ok, instance} = Instance.set_reachable(instance.host)
+      refute instance.unreachable_since
+    end
+
+    test "keeps nil `unreachable_since` of existing matching Instance record having nil `unreachable_since`" do
+      instance = insert(:instance, unreachable_since: nil)
+
+      assert {:ok, instance} = Instance.set_reachable(instance.host)
+      refute instance.unreachable_since
+    end
+
+    test "does NOT create an Instance record in case of no existing matching record" do
+      host = "domain.org"
+      assert nil == Instance.set_reachable(host)
+
+      assert [] = Repo.all(Ecto.Query.from(i in Instance))
+      assert Instance.reachable?(host)
+    end
+  end
+
+  describe "set_unreachable/1" do
+    test "creates new record having `unreachable_since` to current time if record does not exist" do
+      assert {:ok, instance} = Instance.set_unreachable("https://domain.com/path")
+
+      instance = Repo.get(Instance, instance.id)
+      assert instance.unreachable_since
+      assert "domain.com" == instance.host
+    end
+
+    test "sets `unreachable_since` of existing record having nil `unreachable_since`" do
+      instance = insert(:instance, unreachable_since: nil)
+      refute instance.unreachable_since
+
+      assert {:ok, _} = Instance.set_unreachable(instance.host)
+
+      instance = Repo.get(Instance, instance.id)
+      assert instance.unreachable_since
+    end
+
+    test "does NOT modify `unreachable_since` value of existing record in case it's present" do
+      instance =
+        insert(:instance, unreachable_since: NaiveDateTime.add(NaiveDateTime.utc_now(), -10))
+
+      assert instance.unreachable_since
+      initial_value = instance.unreachable_since
+
+      assert {:ok, _} = Instance.set_unreachable(instance.host)
+
+      instance = Repo.get(Instance, instance.id)
+      assert initial_value == instance.unreachable_since
+    end
+  end
+
+  describe "set_unreachable/2" do
+    test "sets `unreachable_since` value of existing record in case it's newer than supplied value" do
+      instance =
+        insert(:instance, unreachable_since: NaiveDateTime.add(NaiveDateTime.utc_now(), -10))
+
+      assert instance.unreachable_since
+
+      past_value = NaiveDateTime.add(NaiveDateTime.utc_now(), -100)
+      assert {:ok, _} = Instance.set_unreachable(instance.host, past_value)
+
+      instance = Repo.get(Instance, instance.id)
+      assert past_value == instance.unreachable_since
+    end
+
+    test "does NOT modify `unreachable_since` value of existing record in case it's equal to or older than supplied value" do
+      instance =
+        insert(:instance, unreachable_since: NaiveDateTime.add(NaiveDateTime.utc_now(), -10))
+
+      assert instance.unreachable_since
+      initial_value = instance.unreachable_since
+
+      assert {:ok, _} = Instance.set_unreachable(instance.host, NaiveDateTime.utc_now())
+
+      instance = Repo.get(Instance, instance.id)
+      assert initial_value == instance.unreachable_since
+    end
+  end
+end

test/web/instances/instances_test.exs

@@ -0,0 +1,112 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2018 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.InstancesTest do
+  alias Pleroma.Instances
+
+  use Pleroma.DataCase
+
+  setup_all do
+    config_path = [:instance, :federation_reachability_timeout_days]
+    initial_setting = Pleroma.Config.get(config_path)
+
+    Pleroma.Config.put(config_path, 1)
+    on_exit(fn -> Pleroma.Config.put(config_path, initial_setting) end)
+
+    :ok
+  end
+
+  describe "reachable?/1" do
+    test "returns `true` for host / url with unknown reachability status" do
+      assert Instances.reachable?("unknown.site")
+      assert Instances.reachable?("http://unknown.site")
+    end
+
+    test "returns `false` for host / url marked unreachable for at least `reachability_datetime_threshold()`" do
+      host = "consistently-unreachable.name"
+      Instances.set_consistently_unreachable(host)
+
+      refute Instances.reachable?(host)
+      refute Instances.reachable?("http://#{host}/path")
+    end
+
+    test "returns `true` for host / url marked unreachable for less than `reachability_datetime_threshold()`" do
+      url = "http://eventually-unreachable.name/path"
+
+      Instances.set_unreachable(url)
+
+      assert Instances.reachable?(url)
+      assert Instances.reachable?(URI.parse(url).host)
+    end
+
+    test "returns true on non-binary input" do
+      assert Instances.reachable?(nil)
+      assert Instances.reachable?(1)
+    end
+  end
+
+  describe "filter_reachable/1" do
+    test "keeps only reachable elements of supplied list" do
+      host = "consistently-unreachable.name"
+      url1 = "http://eventually-unreachable.com/path"
+      url2 = "http://domain.com/path"
+
+      Instances.set_consistently_unreachable(host)
+      Instances.set_unreachable(url1)
+
+      assert [url1, url2] == Instances.filter_reachable([host, url1, url2])
+    end
+  end
+
+  describe "set_reachable/1" do
+    test "sets unreachable url or host reachable" do
+      host = "domain.com"
+      Instances.set_consistently_unreachable(host)
+      refute Instances.reachable?(host)
+
+      Instances.set_reachable(host)
+      assert Instances.reachable?(host)
+    end
+
+    test "keeps reachable url or host reachable" do
+      url = "https://site.name?q="
+      assert Instances.reachable?(url)
+
+      Instances.set_reachable(url)
+      assert Instances.reachable?(url)
+    end
+
+    test "returns error status on non-binary input" do
+      assert {:error, _} = Instances.set_reachable(nil)
+      assert {:error, _} = Instances.set_reachable(1)
+    end
+  end
+
+  # Note: implementation-specific (e.g. Instance) details of set_unreachable/1 should be tested in implementation-specific tests
+  describe "set_unreachable/1" do
+    test "returns error status on non-binary input" do
+      assert {:error, _} = Instances.set_unreachable(nil)
+      assert {:error, _} = Instances.set_unreachable(1)
+    end
+  end
+
+  describe "set_consistently_unreachable/1" do
+    test "sets reachable url or host unreachable" do
+      url = "http://domain.com?q="
+      assert Instances.reachable?(url)
+
+      Instances.set_consistently_unreachable(url)
+      refute Instances.reachable?(url)
+    end
+
+    test "keeps unreachable url or host unreachable" do
+      host = "site.name"
+      Instances.set_consistently_unreachable(host)
+      refute Instances.reachable?(host)
+
+      Instances.set_consistently_unreachable(host)
+      refute Instances.reachable?(host)
+    end
+  end
+end

test/web/ostatus/incoming_documents/delete_handling_test.exs

@@ -2,9 +2,16 @@ defmodule Pleroma.Web.OStatus.DeleteHandlingTest do
   use Pleroma.DataCase
 
   import Pleroma.Factory
+  import Tesla.Mock
+
   alias Pleroma.{Repo, Activity, Object}
   alias Pleroma.Web.OStatus
 
+  setup do
+    mock(fn env -> apply(HttpRequestMock, :request, [env]) end)
+    :ok
+  end
+
   describe "deletions" do
     test "it removes the mentioned activity" do
       note = insert(:note_activity)

test/web/ostatus/ostatus_controller_test.exs

@@ -14,49 +14,51 @@ defmodule Pleroma.Web.OStatus.OStatusControllerTest do
     :ok
   end
 
-  test "decodes a salmon", %{conn: conn} do
+  describe "salmon_incoming" do
-    user = insert(:user)
+    test "decodes a salmon", %{conn: conn} do
-    salmon = File.read!("test/fixtures/salmon.xml")
+      user = insert(:user)
+      salmon = File.read!("test/fixtures/salmon.xml")
 
-    conn =
+      conn =
-      conn
+        conn
-      |> put_req_header("content-type", "application/atom+xml")
+        |> put_req_header("content-type", "application/atom+xml")
-      |> post("/users/#{user.nickname}/salmon", salmon)
+        |> post("/users/#{user.nickname}/salmon", salmon)
 
-    assert response(conn, 200)
+      assert response(conn, 200)
-  end
+    end
 
-  test "decodes a salmon with a changed magic key", %{conn: conn} do
+    test "decodes a salmon with a changed magic key", %{conn: conn} do
-    user = insert(:user)
+      user = insert(:user)
-    salmon = File.read!("test/fixtures/salmon.xml")
+      salmon = File.read!("test/fixtures/salmon.xml")
 
-    conn =
+      conn =
-      conn
+        conn
-      |> put_req_header("content-type", "application/atom+xml")
+        |> put_req_header("content-type", "application/atom+xml")
-      |> post("/users/#{user.nickname}/salmon", salmon)
+        |> post("/users/#{user.nickname}/salmon", salmon)
 
-    assert response(conn, 200)
+      assert response(conn, 200)
 
-    # Set a wrong magic-key for a user so it has to refetch
+      # Set a wrong magic-key for a user so it has to refetch
-    salmon_user = User.get_by_ap_id("http://gs.example.org:4040/index.php/user/1")
+      salmon_user = User.get_by_ap_id("http://gs.example.org:4040/index.php/user/1")
-    # Wrong key
+      # Wrong key
-    info_cng =
+      info_cng =
-      User.Info.remote_user_creation(salmon_user.info, %{
+        User.Info.remote_user_creation(salmon_user.info, %{
-        magic_key:
+          magic_key:
-          "RSA.pu0s-halox4tu7wmES1FVSx6u-4wc0YrUFXcqWXZG4-27UmbCOpMQftRCldNRfyA-qLbz-eqiwrong1EwUvjsD4cYbAHNGHwTvDOyx5AKthQUP44ykPv7kjKGh3DWKySJvcs9tlUG87hlo7AvnMo9pwRS_Zz2CacQ-MKaXyDepk=.AQAB"
+            "RSA.pu0s-halox4tu7wmES1FVSx6u-4wc0YrUFXcqWXZG4-27UmbCOpMQftRCldNRfyA-qLbz-eqiwrong1EwUvjsD4cYbAHNGHwTvDOyx5AKthQUP44ykPv7kjKGh3DWKySJvcs9tlUG87hlo7AvnMo9pwRS_Zz2CacQ-MKaXyDepk=.AQAB"
-      })
+        })
 
-    salmon_user
+      salmon_user
-    |> Ecto.Changeset.change()
+      |> Ecto.Changeset.change()
-    |> Ecto.Changeset.put_embed(:info, info_cng)
+      |> Ecto.Changeset.put_embed(:info, info_cng)
-    |> Repo.update()
+      |> Repo.update()
 
-    conn =
+      conn =
-      build_conn()
+        build_conn()
-      |> put_req_header("content-type", "application/atom+xml")
+        |> put_req_header("content-type", "application/atom+xml")
-      |> post("/users/#{user.nickname}/salmon", salmon)
+        |> post("/users/#{user.nickname}/salmon", salmon)
 
-    assert response(conn, 200)
+      assert response(conn, 200)
+    end
   end
 
   test "gets a feed", %{conn: conn} do

test/web/ostatus/ostatus_test.exs

@@ -6,7 +6,7 @@ defmodule Pleroma.Web.OStatusTest do
   use Pleroma.DataCase
   alias Pleroma.Web.OStatus
   alias Pleroma.Web.XML
-  alias Pleroma.{Object, Repo, User, Activity}
+  alias Pleroma.{Object, Repo, User, Activity, Instances}
   import Pleroma.Factory
   import ExUnit.CaptureLog
 
@@ -311,6 +311,22 @@ test "handle incoming unfollows with existing follow" do
     refute User.following?(follower, followed)
   end
 
+  test "it clears `unreachable` federation status of the sender" do
+    incoming_reaction_xml = File.read!("test/fixtures/share-gs.xml")
+    doc = XML.parse_document(incoming_reaction_xml)
+    actor_uri = XML.string_from_xpath("//author/uri[1]", doc)
+    reacted_to_author_uri = XML.string_from_xpath("//author/uri[2]", doc)
+
+    Instances.set_consistently_unreachable(actor_uri)
+    Instances.set_consistently_unreachable(reacted_to_author_uri)
+    refute Instances.reachable?(actor_uri)
+    refute Instances.reachable?(reacted_to_author_uri)
+
+    {:ok, _} = OStatus.handle_incoming(incoming_reaction_xml)
+    assert Instances.reachable?(actor_uri)
+    refute Instances.reachable?(reacted_to_author_uri)
+  end
+
   describe "new remote user creation" do
     test "returns local users" do
       local_user = insert(:user)

test/web/websub/websub_controller_test.exs

@@ -50,35 +50,37 @@ test "websub subscription confirmation", %{conn: conn} do
     assert_in_delta NaiveDateTime.diff(websub.valid_until, NaiveDateTime.utc_now()), 100, 5
   end
 
-  test "handles incoming feed updates", %{conn: conn} do
+  describe "websub_incoming" do
-    websub = insert(:websub_client_subscription)
+    test "handles incoming feed updates", %{conn: conn} do
-    doc = "some stuff"
+      websub = insert(:websub_client_subscription)
-    signature = Websub.sign(websub.secret, doc)
+      doc = "some stuff"
+      signature = Websub.sign(websub.secret, doc)
 
-    conn =
+      conn =
-      conn
+        conn
-      |> put_req_header("x-hub-signature", "sha1=" <> signature)
+        |> put_req_header("x-hub-signature", "sha1=" <> signature)
-      |> put_req_header("content-type", "application/atom+xml")
+        |> put_req_header("content-type", "application/atom+xml")
-      |> post("/push/subscriptions/#{websub.id}", doc)
+        |> post("/push/subscriptions/#{websub.id}", doc)
 
-    assert response(conn, 200) == "OK"
+      assert response(conn, 200) == "OK"
 
-    assert length(Repo.all(Activity)) == 1
+      assert length(Repo.all(Activity)) == 1
-  end
+    end
 
-  test "rejects incoming feed updates with the wrong signature", %{conn: conn} do
+    test "rejects incoming feed updates with the wrong signature", %{conn: conn} do
-    websub = insert(:websub_client_subscription)
+      websub = insert(:websub_client_subscription)
-    doc = "some stuff"
+      doc = "some stuff"
-    signature = Websub.sign("wrong secret", doc)
+      signature = Websub.sign("wrong secret", doc)
 
-    conn =
+      conn =
-      conn
+        conn
-      |> put_req_header("x-hub-signature", "sha1=" <> signature)
+        |> put_req_header("x-hub-signature", "sha1=" <> signature)
-      |> put_req_header("content-type", "application/atom+xml")
+        |> put_req_header("content-type", "application/atom+xml")
-      |> post("/push/subscriptions/#{websub.id}", doc)
+        |> post("/push/subscriptions/#{websub.id}", doc)
 
-    assert response(conn, 500) == "Error"
+      assert response(conn, 500) == "Error"
 
-    assert length(Repo.all(Activity)) == 0
+      assert length(Repo.all(Activity)) == 0
+    end
   end
 end