Merge remote-tracking branch 'origin/develop' into matrix

PubSub to a channel that will tell us about new chat messages.

lib/pleroma/application.ex

@@ -108,7 +108,10 @@ def start(_type, _args) do
         task_children(@mix_env) ++
         dont_run_in_test(@mix_env) ++
         shout_child(shout_enabled?()) ++
-        [Pleroma.Gopher.Server]
+        [
+          Pleroma.Gopher.Server,
+          {Phoenix.PubSub.PG2, name: :matrix}
+        ]
 
     # See http://elixir-lang.org/docs/stable/elixir/Supervisor.html
     # for other strategies and supported options
@@ -194,6 +197,7 @@ defp cachex_children do
       build_cachex("emoji_packs", expiration: emoji_packs_expiration(), limit: 10),
       build_cachex("failed_proxy_url", limit: 2500),
       build_cachex("banned_urls", default_ttl: :timer.hours(24 * 30), limit: 5_000),
+      build_cachex("matrix", default_ttl: :timer.hours(120), limit: 5000),
       build_cachex("chat_message_id_idempotency_key",
         expiration: chat_message_id_idempotency_key_expiration(),
         limit: 500_000

lib/pleroma/web/activity_pub/side_effects.ex

@@ -518,6 +518,12 @@ defp send_streamables(meta) do
     Keyword.get(meta, :streamables, [])
     |> Enum.each(fn {topics, items} ->
       Streamer.stream(topics, items)
+
+      # Tell the matrix controller that a new message is there so it can
+      # start fetching.
+      with {%User{id: user_id}, %MessageReference{}} <- items do
+        Phoenix.PubSub.broadcast(:matrix, "user:#{user_id}", :chat)
+      end
     end)
 
     meta

lib/pleroma/web/matrix_controller.ex

@@ -0,0 +1,577 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.MatrixController do
+  use Pleroma.Web, :controller
+
+  # alias Pleroma.Web.MediaProxy
+  alias Pleroma.Chat
+  alias Pleroma.Chat.MessageReference
+  alias Pleroma.HTML
+  alias Pleroma.Plugs.AuthenticationPlug
+  alias Pleroma.Plugs.OAuthScopesPlug
+  alias Pleroma.Repo
+  alias Pleroma.User
+  alias Pleroma.Web.MediaProxy
+  alias Pleroma.Web.OAuth.App
+  alias Pleroma.Web.OAuth.Token
+  alias Pleroma.Web.OAuth.Token
+  alias Pleroma.Web.CommonAPI
+  alias Pleroma.Object
+  import Ecto.Query
+
+  plug(
+    OAuthScopesPlug,
+    %{scopes: ["write"]}
+    when action in [
+           :set_presence_status,
+           :set_filter,
+           :send_event,
+           :set_read_marker,
+           :typing,
+           :set_account_data
+         ]
+  )
+
+  plug(
+    OAuthScopesPlug,
+    %{scopes: ["read"]}
+    when action in [
+           :pushrules,
+           :sync,
+           :filter,
+           :key_query,
+           :profile,
+           :joined_groups,
+           :room_keys_version,
+           :key_upload,
+           :capabilities,
+           :room_members,
+           :publicised_groups,
+           :turn_server,
+           :room_messages
+         ]
+  )
+
+  def mxc(url) do
+    "mxc://localhost/#{url |> Base.encode64()}"
+  end
+
+  def client_versions(conn, _) do
+    data = %{
+      versions: ["r0.0.1", "r0.1.0", "r0.2.0", "r0.3.0", "r0.4.0", "r0.5.0"]
+    }
+
+    conn
+    |> json(data)
+  end
+
+  def login_info(conn, _) do
+    data = %{
+      flows: [
+        %{type: "m.login.password"}
+      ]
+    }
+
+    conn
+    |> json(data)
+  end
+
+  def login(conn, params) do
+    IO.inspect(params)
+    username = params["identifier"]["user"] || params["user"]
+    password = params["password"]
+
+    dn = params["initial_device_display_name"]
+
+    with %User{} = user <- User.get_by_nickname(username),
+         true <- AuthenticationPlug.checkpw(password, user.password_hash),
+         {:ok, app} <-
+           App.create(%{client_name: dn, scopes: ~w(read write), redirect_uris: "nowhere"}),
+         {:ok, token} <- Token.create_token(app, user) do
+      data = %{
+        user_id: "@#{user.nickname}:#{Pleroma.Web.Endpoint.host()}",
+        access_token: token.token,
+        device_id: app.client_id,
+        home_server: "matrixtest.ngrok.io"
+      }
+
+      conn
+      |> put_status(200)
+      |> json(data)
+    else
+      _ ->
+        data = %{
+          errcode: "M_FORBIDDEN",
+          error: "Invalid password"
+        }
+
+        conn
+        |> put_status(403)
+        |> json(data)
+    end
+  end
+
+  def presence_status(conn, _) do
+    data = %{
+      presence: "online"
+    }
+
+    conn
+    |> json(data)
+  end
+
+  def set_presence_status(conn, _params) do
+    conn
+    |> json(%{})
+  end
+
+  def pushrules(conn, _) do
+    data = %{
+      global: %{}
+    }
+
+    conn
+    |> json(data)
+  end
+
+  def set_filter(conn, params) do
+    filter_id = :crypto.strong_rand_bytes(32) |> Base.url_encode64(padding: false)
+    Cachex.put(:matrix_cache, "filter:#{filter_id}", params)
+
+    data = %{
+      filter_id: filter_id
+    }
+
+    conn
+    |> json(data)
+  end
+
+  def filter(conn, params) do
+    {:ok, result} = Cachex.get(:matrix_cache, "filter:#{params["filter_id"]}")
+
+    conn
+    |> put_status(200)
+    |> json(result || %{})
+  end
+
+  defp matrix_name(%{local: true, nickname: nick}) do
+    "@#{nick}:#{Pleroma.Web.Endpoint.host()}"
+  end
+
+  defp matrix_name(%{nickname: nick}) do
+    nick =
+      nick
+      |> String.replace("@", ":")
+
+    "@" <> nick
+  end
+
+  defp messages_from_cmrs(cmrs) do
+    cmrs
+    |> Enum.map(fn message ->
+      chat_data = message.object.data
+      author = User.get_cached_by_ap_id(chat_data["actor"])
+
+      {:ok, date, _} = DateTime.from_iso8601(chat_data["published"])
+      {:ok, txn_id} = Cachex.get(:matrix_cache, "txn_id:#{message.id}")
+
+      messages = [
+        %{
+          content: %{
+            body: chat_data["content"] |> HTML.strip_tags(),
+            msgtype: "m.text",
+            format: "org.matrix.custom.html",
+            formatted_body: chat_data["content"]
+          },
+          type: "m.room.message",
+          event_id: message.id,
+          room_id: message.chat_id,
+          sender: matrix_name(author),
+          origin_server_ts: date |> DateTime.to_unix(:millisecond),
+          unsigned: %{
+            age: DateTime.diff(DateTime.utc_now(), date, :millisecond),
+            transaction_id: txn_id
+          }
+        }
+      ]
+
+      messages =
+        if attachment = chat_data["attachment"] do
+          attachment =
+            Pleroma.Web.MastodonAPI.StatusView.render("attachment.json",
+              attachment: attachment
+            )
+
+          att = %{
+            content: %{
+              body: "an image",
+              msgtype: "m.image",
+              url: mxc(attachment.url),
+              info: %{
+                h: 640,
+                w: 480,
+                size: 500_000,
+                mimetype: attachment.pleroma.mime_type
+              }
+            },
+            type: "m.room.message",
+            event_id: attachment.id,
+            room_id: message.chat_id,
+            sender: matrix_name(author),
+            origin_server_ts: date |> DateTime.to_unix(:millisecond),
+            unsigned: %{
+              age: DateTime.diff(DateTime.utc_now(), date, :millisecond)
+            }
+          }
+
+          [att | messages]
+        else
+          messages
+        end
+
+      messages
+    end)
+    |> List.flatten()
+    |> Enum.reverse()
+  end
+
+  def sync(%{assigns: %{user: user}} = conn, params) do
+    with {:ok, timeout} when not is_nil(timeout) <- Ecto.Type.cast(:integer, params["timeout"]) do
+      Phoenix.PubSub.subscribe(:matrix, "user:#{user.id}")
+
+      receive do
+        _ -> :ok
+      after
+        timeout ->
+          :timeout
+      end
+    end
+
+    blocked_ap_ids = User.blocked_users_ap_ids(user)
+
+    user_id = user.id
+
+    chats =
+      from(c in Chat,
+        where: c.user_id == ^user_id,
+        where: c.recipient not in ^blocked_ap_ids,
+        order_by: [desc: c.updated_at]
+      )
+      |> Repo.all()
+      |> Enum.reduce(%{}, fn chat, acc ->
+        recipient = User.get_by_ap_id(chat.recipient)
+
+        membership_events =
+          [user, recipient]
+          |> membership_events_from_list(chat)
+
+        q =
+          chat
+          |> MessageReference.for_chat_query()
+
+        q =
+          if since = params["since"] do
+            from(mr in q,
+              where: mr.id > ^since
+            )
+          else
+            q
+          end
+
+        messages =
+          q
+          |> Repo.all()
+          |> messages_from_cmrs
+
+        room = %{
+          chat.id => %{
+            summary: %{
+              "m.heroes" => [matrix_name(recipient)],
+              "m.joined_member_count" => 2,
+              "m.invited_member_count" => 0
+            },
+            state: %{events: membership_events},
+            ephemeral: %{events: []},
+            timeline: %{
+              events: messages,
+              limited: false,
+              prev_batch: "prev"
+            },
+            account_data: %{events: []},
+            unread_notifications: %{
+              highlight_count: 0,
+              notification_count: 0
+            }
+          }
+        }
+
+        Map.merge(acc, room)
+      end)
+
+    most_recent_cmr_id =
+      Enum.reduce(chats, nil, fn {_k, chat}, acc ->
+        id = List.last(chat.timeline.events)[:event_id]
+
+        if !acc || (acc && acc < id) do
+          id
+        else
+          acc
+        end
+      end)
+
+    data = %{
+      next_batch: most_recent_cmr_id,
+      rooms: %{
+        join: chats,
+        invite: %{},
+        leave: %{}
+      },
+      account_data: %{
+        events: [
+          %{
+            type: "m.direct",
+            content: %{
+              matrix_name(user) => Map.keys(chats)
+            }
+          }
+        ]
+      },
+      presence: %{
+        events: []
+      },
+      to_device: %{
+        events: []
+      },
+      device_one_time_keys_count: %{},
+      device_lists: %{
+        left: [],
+        changed: []
+      }
+    }
+
+    conn
+    |> json(data)
+  end
+
+  def key_query(conn, params) do
+    conn
+    |> json(params)
+  end
+
+  defp nickname_from_matrix_id(mid) do
+    mid
+    |> String.trim_leading("@")
+    |> String.replace(":", "@")
+    |> String.trim_trailing("@#{Pleroma.Web.Endpoint.host()}")
+  end
+
+  def profile(conn, params) do
+    nickname =
+      params["user_id"]
+      |> nickname_from_matrix_id()
+
+    user = User.get_by_nickname(nickname)
+    avatar = User.avatar_url(user) |> MediaProxy.url()
+
+    data = %{
+      displayname: user.name,
+      avatar_url: mxc(avatar)
+    }
+
+    conn
+    |> json(data)
+  end
+
+  def download(conn, params) do
+    {:ok, url} = params["file"] |> Base.decode64()
+
+    # This is stupid
+    with {:ok, %{status: 200} = env} = Pleroma.HTTP.get(url) do
+      conn
+      |> send_resp(200, env.body)
+    end
+  end
+
+  # Not documented, guessing what's expected here
+  def joined_groups(conn, _) do
+    data = %{
+      groups: []
+    }
+
+    conn
+    |> json(data)
+  end
+
+  # Not documented either lololo let's 404
+  def room_keys_version(conn, _) do
+    conn
+    |> put_status(404)
+    |> json("Not found")
+  end
+
+  # let's just pretend this worked.
+  def key_upload(conn, _params) do
+    # Enormous numbers so the client will stop trying to upload more
+    data = %{
+      one_time_key_counts: %{
+        curve25519: 100_000,
+        signed_curve25519: 2_000_000
+      }
+    }
+
+    conn
+    |> put_status(200)
+    |> json(data)
+  end
+
+  def capabilities(conn, _) do
+    data = %{
+      capabilities: %{
+        "m.change_password": %{
+          enabled: false
+        },
+        "m.room_versions": %{
+          default: "1",
+          available: %{
+            "1" => "stable"
+          }
+        }
+      }
+    }
+
+    conn
+    |> json(data)
+  end
+
+  # Just pretend it worked
+  def set_read_marker(%{assigns: %{user: %{id: user_id}}} = conn, %{
+        "m.fully_read" => read_up_to,
+        "room_id" => chat_id
+      }) do
+    with %Chat{user_id: ^user_id} = chat <- Chat.get_by_id(chat_id) do
+      MessageReference.set_all_seen_for_chat(chat, read_up_to)
+    end
+
+    conn
+    |> json(%{})
+  end
+
+  def room_members(%{assigns: %{user: %{id: user_id} = user}} = conn, %{"room_id" => chat_id}) do
+    with %Chat{user_id: ^user_id, recipient: recipient_id} = chat <- Chat.get_by_id(chat_id),
+         %User{} = recipient <- User.get_cached_by_ap_id(recipient_id) do
+      membership_events =
+        [user, recipient]
+        |> membership_events_from_list(chat)
+
+      data = %{
+        chunk: membership_events
+      }
+
+      conn
+      |> json(data)
+    end
+  end
+
+  # Undocumented
+  def publicised_groups(conn, _) do
+    data = %{
+      groups: %{}
+    }
+
+    conn
+    |> json(data)
+  end
+
+  defp membership_events_from_list(users, chat) do
+    users
+    |> Enum.map(fn member ->
+      avatar = User.avatar_url(member) |> MediaProxy.url()
+
+      %{
+        content: %{
+          membership: "join",
+          avatar_url: mxc(avatar),
+          displayname: member.name
+        },
+        type: "m.room.member",
+        event_id: "#{chat.id}/join/#{member.id}",
+        room_id: chat.id,
+        sender: matrix_name(member),
+        origin_ts: DateTime.utc_now() |> DateTime.to_unix(),
+        state_key: matrix_name(member)
+      }
+    end)
+  end
+
+  def turn_server(conn, _) do
+    conn
+    |> put_status(404)
+    |> json("not found")
+  end
+
+  def send_event(
+        %{assigns: %{user: %{id: user_id} = user}} = conn,
+        %{
+          "msgtype" => "m.text",
+          "body" => body,
+          "room_id" => chat_id,
+          "event_type" => "m.room.message",
+          "txn_id" => txn_id
+        }
+      ) do
+    with %Chat{user_id: ^user_id, recipient: recipient_id} = chat <- Chat.get_by_id(chat_id),
+         %User{} = recipient <- User.get_cached_by_ap_id(recipient_id),
+         {:ok, activity} <- CommonAPI.post_chat_message(user, recipient, body) do
+      object = Object.normalize(activity, false)
+      cmr = MessageReference.for_chat_and_object(chat, object)
+
+      # Hard to believe, but element (web) does not use the event id to figure out
+      # if an event returned via sync is the same as the event we send off, but 
+      # instead it uses this transaction id, so if we don't save this (for a
+      # little while) we get doubled messages in the frontend.
+      Cachex.put(:matrix_cache, "txn_id:#{cmr.id}", txn_id)
+
+      data = %{
+        event_id: cmr.id
+      }
+
+      conn
+      |> json(data)
+    end
+  end
+
+  def wellknown(conn, _params) do
+    conn
+    |> put_status(404)
+    |> json("not found")
+  end
+
+  def typing(conn, _) do
+    conn
+    |> json(%{})
+  end
+
+  def set_account_data(conn, _) do
+    conn
+    |> json(%{})
+  end
+
+  def room_messages(%{assigns: %{user: %{id: user_id} = _user}} = conn, %{"room_id" => chat_id}) do
+    with %Chat{user_id: ^user_id} = chat <- Chat.get_by_id(chat_id) do
+      cmrs =
+        chat
+        |> MessageReference.for_chat_query()
+        |> limit(30)
+        |> Repo.all()
+
+      messages =
+        cmrs
+        |> messages_from_cmrs()
+
+      conn
+      |> json(%{chunk: messages, start: List.first(cmrs).id, end: List.last(cmrs).id})
+    end
+  end
+end

lib/pleroma/web/plugs/frontend_static.ex

@@ -33,6 +33,9 @@ def init(opts) do
     |> Map.put(:frontend_type, opts[:frontend_type])
   end
 
+  def call(%{path_info: ["_matrix" | _]} = conn, _opts), do: conn
+  def call(%{path_info: ["api" | _]} = conn, _opts), do: conn
+
   def call(conn, opts) do
     with false <- api_route?(conn.path_info),
          false <- invalid_path?(conn.path_info),

lib/pleroma/web/router.ex

@@ -838,6 +838,48 @@ defmodule Pleroma.Web.Router do
     end
   end
 
+  get("/.well-known/matrix/client", Pleroma.Web.MatrixController, :wellknown)
+
+  scope "/_matrix", Pleroma.Web do
+    pipe_through(:api)
+    get("/client/versions", MatrixController, :client_versions)
+    get("/client/r0/login", MatrixController, :login_info)
+    post("/client/r0/login", MatrixController, :login)
+    get("/client/r0/presence/:user_id/status", MatrixController, :presence_status)
+    get("/client/r0/user/:user_id/filter/:filter_id", MatrixController, :filter)
+    get("/media/r0/download/:authority/:file", MatrixController, :download)
+    get("/media/r0/thumbnail/:authority/:file", MatrixController, :download)
+    # Says it's r0 in the documentation, seems it's actually v1 on iOS
+    get("/media/v1/download/:authority/:file", MatrixController, :download)
+    get("/media/v1/thumbnail/:authority/:file", MatrixController, :download)
+  end
+
+  scope "/_matrix", Pleroma.Web do
+    pipe_through(:authenticated_api)
+    put("/client/r0/presence/:user_id/status", MatrixController, :set_presence_status)
+    get("/client/r0/pushrules", MatrixController, :pushrules)
+    post("/client/r0/user/:user_id/filter", MatrixController, :set_filter)
+    get("/client/r0/sync", MatrixController, :sync)
+    post("/client/r0/keys/query", MatrixController, :key_query)
+    get("/client/r0/profile/:user_id", MatrixController, :profile)
+    get("/client/r0/profile/:user_id/displayname", MatrixController, :profile)
+    get("/client/r0/profile/:user_id/avatar_url", MatrixController, :profile)
+    get("/client/r0/joined_groups", MatrixController, :joined_groups)
+    get("/client/unstable/room_keys/version", MatrixController, :room_keys_version)
+    post("/client/r0/keys/upload", MatrixController, :key_upload)
+    # The iOS client uses this call. No idea what it is about.
+    post("/client/r0/keys/upload/:whoknows", MatrixController, :key_upload)
+    get("/client/r0/capabilities", MatrixController, :capabilities)
+    post("/client/r0/rooms/:room_id/read_markers", MatrixController, :set_read_marker)
+    put("/client/r0/rooms/:room_id/typing/:user_id", MatrixController, :typing)
+    get("/client/r0/rooms/:room_id/members", MatrixController, :room_members)
+    get("/client/r0/rooms/:room_id/messages", MatrixController, :room_messages)
+    post("/client/r0/publicised_groups", MatrixController, :publicised_groups)
+    get("/client/r0/voip/turnServer", MatrixController, :turn_server)
+    put("/client/r0/rooms/:room_id/send/:event_type/:txn_id", MatrixController, :send_event)
+    put("/client/r0/user/:user_id/account_data/:type", MatrixController, :set_account_data)
+  end
+
   scope "/", Pleroma.Web.MongooseIM do
     get("/user_exists", MongooseIMController, :user_exists)
     get("/check_password", MongooseIMController, :check_password)

test/pleroma/web/matrix_controller_test.exs

@@ -0,0 +1,143 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.MatrixControllerTest do
+  use Pleroma.Web.ConnCase
+
+  alias Pleroma.Web.CommonAPI
+  alias Pleroma.Chat
+  alias Pleroma.Chat.MessageReference
+  alias Pleroma.Object
+  import Pleroma.Factory
+
+  setup do
+    %{user: user, conn: conn} = oauth_access(["read"])
+    other_user = insert(:user)
+    third_user = insert(:user)
+
+    {:ok, chat_activity} = CommonAPI.post_chat_message(user, other_user, "Hey")
+    {:ok, chat_activity_two} = CommonAPI.post_chat_message(third_user, user, "Henlo")
+
+    chat = Chat.get(user.id, other_user.ap_id)
+    chat_two = Chat.get(user.id, third_user.ap_id)
+
+    chat_object = Object.normalize(chat_activity)
+    cmr = MessageReference.for_chat_and_object(chat, chat_object)
+
+    chat_object_two = Object.normalize(chat_activity_two)
+    cmr_two = MessageReference.for_chat_and_object(chat_two, chat_object_two)
+
+    %{
+      user: user,
+      other_user: other_user,
+      third_user: third_user,
+      conn: conn,
+      chat_activity: chat_activity,
+      chat_activity_two: chat_activity_two,
+      chat: chat,
+      chat_two: chat_two,
+      cmr: cmr,
+      cmr_two: cmr_two
+    }
+  end
+
+  describe "setting read markers" do
+    test "it sets all messages up to a point as read", %{
+      conn: conn,
+      cmr: cmr,
+      chat: chat,
+      user: user,
+      other_user: other_user
+    } do
+      {:ok, chat_activity} = CommonAPI.post_chat_message(other_user, user, "morning weebs 2")
+      chat_object = Object.normalize(chat_activity)
+
+      conn
+      |> post("/_matrix/client/r0/rooms/#{chat.id}/read_markers", %{"m.fully_read": cmr.id})
+
+      cmr_two = MessageReference.for_chat_and_object(chat, chat_object)
+      assert cmr_two.unread
+
+      cmr = MessageReference.get_by_id(cmr.id)
+      refute cmr.unread
+    end
+  end
+
+  describe "sync" do
+    test "without options, it returns all chat messages the user has", %{
+      conn: conn,
+      chat: chat,
+      chat_two: chat_two,
+      cmr: cmr,
+      cmr_two: cmr_two
+    } do
+      %{
+        "next_batch" => next_batch,
+        "rooms" => %{
+          "join" => joined_rooms
+        }
+      } =
+        conn
+        |> get("_matrix/client/r0/sync")
+        |> json_response(200)
+
+      assert chat_room = joined_rooms[chat.id]
+      assert chat_room_two = joined_rooms[chat_two.id]
+
+      assert [message] = chat_room["timeline"]["events"]
+      assert [message_two] = chat_room_two["timeline"]["events"]
+
+      assert message["content"]["formatted_body"] == "Hey"
+      assert message_two["content"]["formatted_body"] == "Henlo"
+
+      assert message["event_id"] == cmr.id
+      assert message_two["event_id"] == cmr_two.id
+
+      # Next batch contains the largest ChatMessageReference id
+
+      assert next_batch == cmr_two.id
+    end
+
+    test "given a `since` option, it only returns chat messages after that point", %{
+      conn: conn,
+      cmr_two: cmr_two,
+      chat: chat,
+      chat_two: chat_two,
+      user: user,
+      other_user: other_user
+    } do
+      {:ok, _} = CommonAPI.post_chat_message(user, other_user, "morning weebs")
+
+      %{
+        "rooms" => %{
+          "join" => joined_rooms
+        }
+      } =
+        conn
+        |> get("_matrix/client/r0/sync?since=#{cmr_two.id}")
+        |> json_response(200)
+
+      assert joined_rooms[chat_two.id]
+      assert chat_room = joined_rooms[chat.id]
+      assert [message] = chat_room["timeline"]["events"]
+      assert message["content"]["body"] == "morning weebs"
+    end
+  end
+
+  describe "room messages" do
+    test "it returns messages", %{
+      conn: conn,
+      chat: chat
+    } do
+      res =
+        conn
+        |> get("_matrix/client/r0/rooms/#{chat.id}/messages")
+        |> json_response(200)
+
+      assert res["chunk"]
+      assert res["start"]
+      assert res["end"]
+    end
+  end
+end