Add outgoing federation

lib/clacks/activitypub/federator.ex

@@ -0,0 +1,92 @@
+defmodule Clacks.ActivityPub.Federator do
+  require Logger
+  alias Clacks.{Repo, Actor, User, Keys}
+  import Ecto.Query
+
+  @public "https://www.w3.org/ns/activitystreams#Public"
+
+  @spec federate_to_followers(activity :: map(), actor :: Actor.t()) :: :ok | {:error, any()}
+  def federate_to_followers(activity, actor) do
+    Repo.all(
+      from a in Actor, where: fragment("?->>'id'", a.data) in ^actor.followers, select: a.data
+    )
+    |> Enum.map(&inbox_for(activity, &1))
+    |> Enum.uniq()
+    |> Enum.reduce_while(:ok, fn inbox, _acc ->
+      case federate(activity, inbox) do
+        {:error, _} = err ->
+          {:halt, err}
+
+        _ ->
+          {:cont, :ok}
+      end
+    end)
+  end
+
+  @spec federate(activity :: map(), inbox :: String.t()) :: :ok | {:error, any()}
+  def federate(%{"actor" => actor_id} = activity, inbox) do
+    Logger.info("Federating #{activity["id"]} to #{inbox}")
+    %{host: inbox_host, path: inbox_path} = URI.parse(inbox)
+
+    {:ok, body} = Jason.encode(activity)
+    digest = "SHA-256=" <> Base.encode64(:crypto.hash(:sha256, body))
+    date = signature_timestamp()
+
+    signature_params = %{
+      "(request-target)": "post #{inbox_path}",
+      host: inbox_host,
+      "content-length": byte_size(body),
+      digest: digest
+    }
+
+    {private_key, key_id} = private_key_for_actor(actor_id)
+    signature_string = HTTPSignatures.sign(private_key, key_id, signature_params)
+
+    headers = [
+      {"Content-Type", "application/activity+json"},
+      {"Date", date},
+      {"Signature", signature_string},
+      {"Digest", digest}
+    ]
+
+    opts = [hackney: Application.get_env(:clacks, :hackney_opts, [])]
+
+    case HTTPoison.post(inbox, body, headers, opts) do
+      {:ok, %HTTPoison.Response{status_code: status_code}} when status_code in 200..299 ->
+        :ok
+
+      {:error, _} = err ->
+        err
+    end
+  end
+
+  # see https://www.w3.org/TR/activitypub/#shared-inbox-delivery
+  @spec inbox_for(activity :: map(), actor :: map()) :: String.t()
+  defp inbox_for(activity, actor) do
+    cond do
+      @public in activity["to"] or @public in activity["cc"] ->
+        shared_inbox_for(actor)
+
+      actor.data["followers"] in activity["to"] or actor.data["followers"] in activity["cc"] ->
+        shared_inbox_for(actor)
+
+      true ->
+        actor.data["inbox"]
+    end
+  end
+
+  @spec shared_inbox_for(actor :: map()) :: String.t()
+  defp shared_inbox_for(%{"endpoints" => %{"sharedInbox" => shared}}), do: shared
+  defp shared_inbox_for(%{"inbox" => inbox}), do: inbox
+
+  @spec signature_timestamp() :: String.t()
+  defp signature_timestamp(date \\ NaiveDateTime.utc_now()) do
+    Timex.format!(date, "{WDshort}, {0D} {Mshort} {YYYY} {h24}:{m}:{s} GMT")
+  end
+
+  defp private_key_for_actor(ap_id) do
+    %Actor{user: %User{private_key: pem}} = Actor.get_by_ap_id(ap_id) |> Repo.preload(:user)
+    {:ok, private_key, _} = Keys.keys_from_private_key_pem(pem)
+    {private_key, ap_id <> "#main-key"}
+  end
+end

mix.exs

@@ -47,7 +47,8 @@ defmodule Clacks.MixProject do
       {:http_signatures,
        git: "https://git.pleroma.social/pleroma/http_signatures.git",
        ref: "293d77bb6f4a67ac8bde1428735c3b42f22cbb30"},
-      {:httpoison, "~> 1.5.1"}
+      {:httpoison, "~> 1.5.1"},
+      {:timex, "~> 3.6.1"}
     ]
   end
 

mix.lock

@@ -1,6 +1,7 @@
 %{
   "base62": {:hex, :base62, "1.2.1", "4866763e08555a7b3917064e9eef9194c41667276c51b59de2bc42c6ea65f806", [:mix], [{:custom_base, "~> 0.2.1", [hex: :custom_base, repo: "hexpm", optional: false]}], "hexpm"},
   "certifi": {:hex, :certifi, "2.5.1", "867ce347f7c7d78563450a18a6a28a8090331e77fa02380b4a21962a65d36ee5", [:rebar3], [{:parse_trans, "~>3.3", [hex: :parse_trans, repo: "hexpm", optional: false]}], "hexpm"},
+  "combine": {:hex, :combine, "0.10.0", "eff8224eeb56498a2af13011d142c5e7997a80c8f5b97c499f84c841032e429f", [:mix], [], "hexpm"},
   "connection": {:hex, :connection, "1.0.4", "a1cae72211f0eef17705aaededacac3eb30e6625b04a6117c1b2db6ace7d5976", [:mix], [], "hexpm"},
   "cowboy": {:hex, :cowboy, "2.6.3", "99aa50e94e685557cad82e704457336a453d4abcb77839ad22dbe71f311fcc06", [:rebar3], [{:cowlib, "~> 2.7.3", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "~> 1.7.1", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm"},
   "cowlib": {:hex, :cowlib, "2.7.3", "a7ffcd0917e6d50b4d5fb28e9e2085a0ceb3c97dea310505f7460ff5ed764ce9", [:rebar3], [], "hexpm"},
@@ -33,5 +34,7 @@
   "ranch": {:hex, :ranch, "1.7.1", "6b1fab51b49196860b733a49c07604465a47bdb78aa10c1c16a3d199f7f8c881", [:rebar3], [], "hexpm"},
   "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.5", "6eaf7ad16cb568bb01753dbbd7a95ff8b91c7979482b95f38443fe2c8852a79b", [:make, :mix, :rebar3], [], "hexpm"},
   "telemetry": {:hex, :telemetry, "0.4.0", "8339bee3fa8b91cb84d14c2935f8ecf399ccd87301ad6da6b71c09553834b2ab", [:rebar3], [], "hexpm"},
+  "timex": {:hex, :timex, "3.6.1", "efdf56d0e67a6b956cc57774353b0329c8ab7726766a11547e529357ffdc1d56", [:mix], [{:combine, "~> 0.10", [hex: :combine, repo: "hexpm", optional: false]}, {:gettext, "~> 0.10", [hex: :gettext, repo: "hexpm", optional: false]}, {:tzdata, "~> 0.1.8 or ~> 0.5 or ~> 1.0.0", [hex: :tzdata, repo: "hexpm", optional: false]}], "hexpm"},
+  "tzdata": {:hex, :tzdata, "1.0.1", "f6027a331af7d837471248e62733c6ebee86a72e57c613aa071ebb1f750fc71a", [:mix], [{:hackney, "~> 1.0", [hex: :hackney, repo: "hexpm", optional: false]}], "hexpm"},
   "unicode_util_compat": {:hex, :unicode_util_compat, "0.4.1", "d869e4c68901dd9531385bb0c8c40444ebf624e60b6962d95952775cac5e90cd", [:rebar3], [], "hexpm"},
 }