Merge branch 'feature/incoming_ostatus' of ssh.gitgud.io:lambadalambda/pleroma into feature/incoming_ostatus

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -36,7 +36,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
       {:ok, activity} = add_conversation_id(activity)
 
       if actor.local do
-        Pleroma.Web.Websub.publish(Pleroma.Web.OStatus.feed_path(actor), actor, activity)
+        Pleroma.Web.Federator.enqueue(:publish, activity)
        end
 
       {:ok, activity}

lib/pleroma/web/federator/federator.ex

@@ -0,0 +1,32 @@
+defmodule Pleroma.Web.Federator do
+  alias Pleroma.User
+  require Logger
+
+  @websub_verifier Application.get_env(:pleroma, :websub_verifier)
+
+  def handle(:publish, activity) do
+    Logger.debug("Running publish for #{activity.data["id"]}")
+    with actor when not is_nil(actor) <- User.get_cached_by_ap_id(activity.data["actor"]) do
+      Pleroma.Web.Websub.publish(Pleroma.Web.OStatus.feed_path(actor), actor, activity)
+    end
+  end
+
+  def handle(:verify_websub, websub) do
+    Logger.debug("Running websub verification for #{websub.id} (#{websub.topic}, #{websub.callback})")
+    @websub_verifier.verify(websub)
+  end
+
+  def handle(type, payload) do
+    Logger.debug("Unknown task: #{type}")
+    {:error, "Don't know what do do with this"}
+  end
+
+  def enqueue(type, payload) do
+    # for now, just run immediately in a new process.
+    if Mix.env == :test do
+      handle(type, payload)
+    else
+      spawn(fn -> handle(type, payload) end)
+    end
+  end
+end

lib/pleroma/web/salmon/salmon.ex

@@ -57,7 +57,7 @@ defmodule Pleroma.Web.Salmon do
     end
   end
 
-  defp decode_key("RSA." <> magickey) do
+  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)
@@ -70,4 +70,58 @@ defmodule Pleroma.Web.Salmon do
 
     {: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 generate_rsa_pem do
+    port = Port.open({:spawn, "openssl genrsa"}, [:binary])
+    {:ok, pem} = receive do
+      {^port, {:data, pem}} -> {:ok, pem}
+    end
+    Port.close(port)
+    if Regex.match?(~r/RSA PRIVATE KEY/, pem) do
+      {:ok, pem}
+    else
+      :error
+    end
+  end
+
+  def keys_from_pem(pem) do
+    [private_key_code] = :public_key.pem_decode(pem)
+    private_key = :public_key.pem_entry_decode(private_key_code)
+    {:RSAPrivateKey, _, modulus, exponent, _, _, _, _, _, _, _} = private_key
+    public_key = {:RSAPublicKey, modulus, exponent}
+    {:ok, private_key, public_key}
+  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 = :public_key.sign(signed_text, :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
 end

lib/pleroma/web/websub/websub.ex

@@ -1,13 +1,11 @@
 defmodule Pleroma.Web.Websub do
   alias Pleroma.Repo
-  alias Pleroma.Web.Websub.WebsubServerSubscription
+  alias Pleroma.Web.Websub.{WebsubServerSubscription, WebsubClientSubscription}
   alias Pleroma.Web.OStatus.FeedRepresenter
   alias Pleroma.Web.OStatus
 
   import Ecto.Query
 
-  @websub_verifier Application.get_env(:pleroma, :websub_verifier)
-
   def verify(subscription, getter \\ &HTTPoison.get/3 ) do
     challenge = Base.encode16(:crypto.strong_rand_bytes(8))
     lease_seconds = NaiveDateTime.diff(subscription.valid_until, subscription.updated_at) |> to_string
@@ -71,8 +69,7 @@ defmodule Pleroma.Web.Websub do
       change = Ecto.Changeset.change(websub, %{valid_until: NaiveDateTime.add(websub.updated_at, lease_time)})
       websub = Repo.update!(change)
 
-      # Just spawn that for now, maybe pool later.
-      spawn(fn -> @websub_verifier.verify(websub) end)
+      Pleroma.Web.Federator.enqueue(:verify_websub, websub)
 
       {:ok, websub}
     else {:error, reason} ->
@@ -99,4 +96,23 @@ defmodule Pleroma.Web.Websub do
       {:error, "Wrong topic requested, expected #{OStatus.feed_path(user)}, got #{topic}"}
     end
   end
+
+  def subscribe(user, topic) do
+    # Race condition, use transactions
+    {:ok, subscription} = with subscription when not is_nil(subscription) <- Repo.get_by(WebsubClientSubscription, topic: topic) do
+      subscribers = [user.ap_id, subscription.subcribers] |> Enum.uniq
+      change = Ecto.Changeset.change(subscription, %{subscribers: subscribers})
+      Repo.update(change)
+    else _e ->
+      subscription = %WebsubClientSubscription{
+        topic: topic,
+        subscribers: [user.ap_id],
+        state: "requested",
+        secret: :crypto.strong_rand_bytes(8) |> Base.url_encode64
+      }
+      Repo.insert(subscription)
+    end
+
+    {:ok, subscription}
+  end
 end

lib/pleroma/web/websub/websub_client_subscription.ex

@@ -0,0 +1,13 @@
+defmodule Pleroma.Web.Websub.WebsubClientSubscription do
+  use Ecto.Schema
+
+  schema "websub_client_subscriptions" do
+    field :topic, :string
+    field :secret, :string
+    field :valid_until, :naive_datetime
+    field :state, :string
+    field :subscribers, {:array, :string}, default: []
+
+    timestamps()
+  end
+end