Merge branch 'develop' of git.pleroma.social:pleroma/pleroma into remake-remodel-dms

lib/pleroma/application.ex

@@ -173,7 +173,14 @@ defmodule Pleroma.Application do
   defp streamer_child(env) when env in [:test, :benchmark], do: []
 
   defp streamer_child(_) do
-    [Pleroma.Web.Streamer.supervisor()]
+    [
+      {Registry,
+       [
+         name: Pleroma.Web.Streamer.registry(),
+         keys: :duplicate,
+         partitions: System.schedulers_online()
+       ]}
+    ]
   end
 
   defp chat_child(_env, true) do

lib/pleroma/mfa.ex

@@ -0,0 +1,156 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.MFA do
+  @moduledoc """
+  The MFA context.
+  """
+
+  alias Comeonin.Pbkdf2
+  alias Pleroma.User
+
+  alias Pleroma.MFA.BackupCodes
+  alias Pleroma.MFA.Changeset
+  alias Pleroma.MFA.Settings
+  alias Pleroma.MFA.TOTP
+
+  @doc """
+  Returns MFA methods the user has enabled.
+
+  ## Examples
+
+    iex> Pleroma.MFA.supported_method(User)
+    "totp, u2f"
+  """
+  @spec supported_methods(User.t()) :: String.t()
+  def supported_methods(user) do
+    settings = fetch_settings(user)
+
+    Settings.mfa_methods()
+    |> Enum.reduce([], fn m, acc ->
+      if method_enabled?(m, settings) do
+        acc ++ [m]
+      else
+        acc
+      end
+    end)
+    |> Enum.join(",")
+  end
+
+  @doc "Checks that user enabled MFA"
+  def require?(user) do
+    fetch_settings(user).enabled
+  end
+
+  @doc """
+  Display MFA settings of user
+  """
+  def mfa_settings(user) do
+    settings = fetch_settings(user)
+
+    Settings.mfa_methods()
+    |> Enum.map(fn m -> [m, method_enabled?(m, settings)] end)
+    |> Enum.into(%{enabled: settings.enabled}, fn [a, b] -> {a, b} end)
+  end
+
+  @doc false
+  def fetch_settings(%User{} = user) do
+    user.multi_factor_authentication_settings || %Settings{}
+  end
+
+  @doc "clears backup codes"
+  def invalidate_backup_code(%User{} = user, hash_code) do
+    %{backup_codes: codes} = fetch_settings(user)
+
+    user
+    |> Changeset.cast_backup_codes(codes -- [hash_code])
+    |> User.update_and_set_cache()
+  end
+
+  @doc "generates backup codes"
+  @spec generate_backup_codes(User.t()) :: {:ok, list(binary)} | {:error, String.t()}
+  def generate_backup_codes(%User{} = user) do
+    with codes <- BackupCodes.generate(),
+         hashed_codes <- Enum.map(codes, &Pbkdf2.hashpwsalt/1),
+         changeset <- Changeset.cast_backup_codes(user, hashed_codes),
+         {:ok, _} <- User.update_and_set_cache(changeset) do
+      {:ok, codes}
+    else
+      {:error, msg} ->
+        %{error: msg}
+    end
+  end
+
+  @doc """
+  Generates secret key and set delivery_type to 'app' for TOTP method.
+  """
+  @spec setup_totp(User.t()) :: {:ok, User.t()} | {:error, Ecto.Changeset.t()}
+  def setup_totp(user) do
+    user
+    |> Changeset.setup_totp(%{secret: TOTP.generate_secret(), delivery_type: "app"})
+    |> User.update_and_set_cache()
+  end
+
+  @doc """
+  Confirms the TOTP method for user.
+
+  `attrs`:
+    `password` - current user password
+    `code` - TOTP token
+  """
+  @spec confirm_totp(User.t(), map()) :: {:ok, User.t()} | {:error, Ecto.Changeset.t() | atom()}
+  def confirm_totp(%User{} = user, attrs) do
+    with settings <- user.multi_factor_authentication_settings.totp,
+         {:ok, :pass} <- TOTP.validate_token(settings.secret, attrs["code"]) do
+      user
+      |> Changeset.confirm_totp()
+      |> User.update_and_set_cache()
+    end
+  end
+
+  @doc """
+  Disables the TOTP method for user.
+
+  `attrs`:
+    `password` - current user password
+  """
+  @spec disable_totp(User.t()) :: {:ok, User.t()} | {:error, Ecto.Changeset.t()}
+  def disable_totp(%User{} = user) do
+    user
+    |> Changeset.disable_totp()
+    |> Changeset.disable()
+    |> User.update_and_set_cache()
+  end
+
+  @doc """
+  Force disables all MFA methods for user.
+  """
+  @spec disable(User.t()) :: {:ok, User.t()} | {:error, Ecto.Changeset.t()}
+  def disable(%User{} = user) do
+    user
+    |> Changeset.disable_totp()
+    |> Changeset.disable(true)
+    |> User.update_and_set_cache()
+  end
+
+  @doc """
+  Checks if the user has MFA method enabled.
+  """
+  def method_enabled?(method, settings) do
+    with {:ok, %{confirmed: true} = _} <- Map.fetch(settings, method) do
+      true
+    else
+      _ -> false
+    end
+  end
+
+  @doc """
+  Checks if the user has enabled at least one MFA method.
+  """
+  def enabled?(settings) do
+    Settings.mfa_methods()
+    |> Enum.map(fn m -> method_enabled?(m, settings) end)
+    |> Enum.any?()
+  end
+end

lib/pleroma/mfa/backup_codes.ex

@@ -0,0 +1,31 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.MFA.BackupCodes do
+  @moduledoc """
+  This module contains functions for generating backup codes.
+  """
+  alias Pleroma.Config
+
+  @config_ns [:instance, :multi_factor_authentication, :backup_codes]
+
+  @doc """
+  Generates backup codes.
+  """
+  @spec generate(Keyword.t()) :: list(String.t())
+  def generate(opts \\ []) do
+    number_of_codes = Keyword.get(opts, :number_of_codes, default_backup_codes_number())
+    code_length = Keyword.get(opts, :length, default_backup_codes_code_length())
+
+    Enum.map(1..number_of_codes, fn _ ->
+      :crypto.strong_rand_bytes(div(code_length, 2))
+      |> Base.encode16(case: :lower)
+    end)
+  end
+
+  defp default_backup_codes_number, do: Config.get(@config_ns ++ [:number], 5)
+
+  defp default_backup_codes_code_length,
+    do: Config.get(@config_ns ++ [:length], 16)
+end

lib/pleroma/mfa/changeset.ex

@@ -0,0 +1,64 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.MFA.Changeset do
+  alias Pleroma.MFA
+  alias Pleroma.MFA.Settings
+  alias Pleroma.User
+
+  def disable(%Ecto.Changeset{} = changeset, force \\ false) do
+    settings =
+      changeset
+      |> Ecto.Changeset.apply_changes()
+      |> MFA.fetch_settings()
+
+    if force || not MFA.enabled?(settings) do
+      put_change(changeset, %Settings{settings | enabled: false})
+    else
+      changeset
+    end
+  end
+
+  def disable_totp(%User{multi_factor_authentication_settings: settings} = user) do
+    user
+    |> put_change(%Settings{settings | totp: %Settings.TOTP{}})
+  end
+
+  def confirm_totp(%User{multi_factor_authentication_settings: settings} = user) do
+    totp_settings = %Settings.TOTP{settings.totp | confirmed: true}
+
+    user
+    |> put_change(%Settings{settings | totp: totp_settings, enabled: true})
+  end
+
+  def setup_totp(%User{} = user, attrs) do
+    mfa_settings = MFA.fetch_settings(user)
+
+    totp_settings =
+      %Settings.TOTP{}
+      |> Ecto.Changeset.cast(attrs, [:secret, :delivery_type])
+
+    user
+    |> put_change(%Settings{mfa_settings | totp: Ecto.Changeset.apply_changes(totp_settings)})
+  end
+
+  def cast_backup_codes(%User{} = user, codes) do
+    user
+    |> put_change(%Settings{
+      user.multi_factor_authentication_settings
+      | backup_codes: codes
+    })
+  end
+
+  defp put_change(%User{} = user, settings) do
+    user
+    |> Ecto.Changeset.change()
+    |> put_change(settings)
+  end
+
+  defp put_change(%Ecto.Changeset{} = changeset, settings) do
+    changeset
+    |> Ecto.Changeset.put_change(:multi_factor_authentication_settings, settings)
+  end
+end

lib/pleroma/mfa/settings.ex

@@ -0,0 +1,24 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.MFA.Settings do
+  use Ecto.Schema
+
+  @primary_key false
+
+  @mfa_methods [:totp]
+  embedded_schema do
+    field(:enabled, :boolean, default: false)
+    field(:backup_codes, {:array, :string}, default: [])
+
+    embeds_one :totp, TOTP, on_replace: :delete, primary_key: false do
+      field(:secret, :string)
+      # app | sms
+      field(:delivery_type, :string, default: "app")
+      field(:confirmed, :boolean, default: false)
+    end
+  end
+
+  def mfa_methods, do: @mfa_methods
+end

lib/pleroma/mfa/token.ex

@@ -0,0 +1,106 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.MFA.Token do
+  use Ecto.Schema
+  import Ecto.Query
+  import Ecto.Changeset
+
+  alias Pleroma.Repo
+  alias Pleroma.User
+  alias Pleroma.Web.OAuth.Authorization
+  alias Pleroma.Web.OAuth.Token, as: OAuthToken
+
+  @expires 300
+
+  schema "mfa_tokens" do
+    field(:token, :string)
+    field(:valid_until, :naive_datetime_usec)
+
+    belongs_to(:user, User, type: FlakeId.Ecto.CompatType)
+    belongs_to(:authorization, Authorization)
+
+    timestamps()
+  end
+
+  def get_by_token(token) do
+    from(
+      t in __MODULE__,
+      where: t.token == ^token,
+      preload: [:user, :authorization]
+    )
+    |> Repo.find_resource()
+  end
+
+  def validate(token) do
+    with {:fetch_token, {:ok, token}} <- {:fetch_token, get_by_token(token)},
+         {:expired, false} <- {:expired, is_expired?(token)} do
+      {:ok, token}
+    else
+      {:expired, _} -> {:error, :expired_token}
+      {:fetch_token, _} -> {:error, :not_found}
+      error -> {:error, error}
+    end
+  end
+
+  def create_token(%User{} = user) do
+    %__MODULE__{}
+    |> change
+    |> assign_user(user)
+    |> put_token
+    |> put_valid_until
+    |> Repo.insert()
+  end
+
+  def create_token(user, authorization) do
+    %__MODULE__{}
+    |> change
+    |> assign_user(user)
+    |> assign_authorization(authorization)
+    |> put_token
+    |> put_valid_until
+    |> Repo.insert()
+  end
+
+  defp assign_user(changeset, user) do
+    changeset
+    |> put_assoc(:user, user)
+    |> validate_required([:user])
+  end
+
+  defp assign_authorization(changeset, authorization) do
+    changeset
+    |> put_assoc(:authorization, authorization)
+    |> validate_required([:authorization])
+  end
+
+  defp put_token(changeset) do
+    changeset
+    |> change(%{token: OAuthToken.Utils.generate_token()})
+    |> validate_required([:token])
+    |> unique_constraint(:token)
+  end
+
+  defp put_valid_until(changeset) do
+    expires_in = NaiveDateTime.add(NaiveDateTime.utc_now(), @expires)
+
+    changeset
+    |> change(%{valid_until: expires_in})
+    |> validate_required([:valid_until])
+  end
+
+  def is_expired?(%__MODULE__{valid_until: valid_until}) do
+    NaiveDateTime.diff(NaiveDateTime.utc_now(), valid_until) > 0
+  end
+
+  def is_expired?(_), do: false
+
+  def delete_expired_tokens do
+    from(
+      q in __MODULE__,
+      where: fragment("?", q.valid_until) < ^Timex.now()
+    )
+    |> Repo.delete_all()
+  end
+end

lib/pleroma/mfa/totp.ex

@@ -0,0 +1,86 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.MFA.TOTP do
+  @moduledoc """
+  This module represents functions to create secrets for
+  TOTP Application as well as validate them with a time based token.
+  """
+  alias Pleroma.Config
+
+  @config_ns [:instance, :multi_factor_authentication, :totp]
+
+  @doc """
+  https://github.com/google/google-authenticator/wiki/Key-Uri-Format
+  """
+  def provisioning_uri(secret, label, opts \\ []) do
+    query =
+      %{
+        secret: secret,
+        issuer: Keyword.get(opts, :issuer, default_issuer()),
+        digits: Keyword.get(opts, :digits, default_digits()),
+        period: Keyword.get(opts, :period, default_period())
+      }
+      |> Enum.filter(fn {_, v} -> not is_nil(v) end)
+      |> Enum.into(%{})
+      |> URI.encode_query()
+
+    %URI{scheme: "otpauth", host: "totp", path: "/" <> label, query: query}
+    |> URI.to_string()
+  end
+
+  defp default_period, do: Config.get(@config_ns ++ [:period])
+  defp default_digits, do: Config.get(@config_ns ++ [:digits])
+
+  defp default_issuer,
+    do: Config.get(@config_ns ++ [:issuer], Config.get([:instance, :name]))
+
+  @doc "Creates a random Base 32 encoded string"
+  def generate_secret do
+    Base.encode32(:crypto.strong_rand_bytes(10))
+  end
+
+  @doc "Generates a valid token based on a secret"
+  def generate_token(secret) do
+    :pot.totp(secret)
+  end
+
+  @doc """
+  Validates a given token based on a secret.
+
+  optional parameters:
+  `token_length` default `6`
+  `interval_length` default `30`
+  `window` default 0
+
+  Returns {:ok, :pass} if the token is valid and
+  {:error, :invalid_token} if it is not.
+  """
+  @spec validate_token(String.t(), String.t()) ::
+          {:ok, :pass} | {:error, :invalid_token | :invalid_secret_and_token}
+  def validate_token(secret, token)
+      when is_binary(secret) and is_binary(token) do
+    opts = [
+      token_length: default_digits(),
+      interval_length: default_period()
+    ]
+
+    validate_token(secret, token, opts)
+  end
+
+  def validate_token(_, _), do: {:error, :invalid_secret_and_token}
+
+  @doc "See `validate_token/2`"
+  @spec validate_token(String.t(), String.t(), Keyword.t()) ::
+          {:ok, :pass} | {:error, :invalid_token | :invalid_secret_and_token}
+  def validate_token(secret, token, options)
+      when is_binary(secret) and is_binary(token) do
+    case :pot.valid_totp(token, secret, options) do
+      true -> {:ok, :pass}
+      false -> {:error, :invalid_token}
+    end
+  end
+
+  def validate_token(_, _, _), do: {:error, :invalid_secret_and_token}
+end

lib/pleroma/plugs/ensure_authenticated_plug.ex

@@ -15,6 +15,20 @@ defmodule Pleroma.Plugs.EnsureAuthenticatedPlug do
   end
 
   @impl true
+  def perform(
+        %{
+          assigns: %{
+            auth_credentials: %{password: _},
+            user: %User{multi_factor_authentication_settings: %{enabled: true}}
+          }
+        } = conn,
+        _
+      ) do
+    conn
+    |> render_error(:forbidden, "Two-factor authentication enabled, you must use a access token.")
+    |> halt()
+  end
+
   def perform(%{assigns: %{user: %User{}}} = conn, _) do
     conn
   end

lib/pleroma/user.ex

@@ -20,6 +20,7 @@ defmodule Pleroma.User do
   alias Pleroma.Formatter
   alias Pleroma.HTML
   alias Pleroma.Keys
+  alias Pleroma.MFA
   alias Pleroma.Notification
   alias Pleroma.Object
   alias Pleroma.Registration
@@ -190,6 +191,12 @@ defmodule Pleroma.User do
     # `:subscribers` is deprecated (replaced with `subscriber_users` relation)
     field(:subscribers, {:array, :string}, default: [])
 
+    embeds_one(
+      :multi_factor_authentication_settings,
+      MFA.Settings,
+      on_replace: :delete
+    )
+
     timestamps()
   end
 
@@ -927,6 +934,7 @@ defmodule Pleroma.User do
     end
   end
 
+  @spec get_by_nickname(String.t()) :: User.t() | nil
   def get_by_nickname(nickname) do
     Repo.get_by(User, nickname: nickname) ||
       if Regex.match?(~r(@#{Pleroma.Web.Endpoint.host()})i, nickname) do

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -177,12 +177,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
       BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})
 
-      Notification.create_notifications(activity)
-
-      conversation = create_or_bump_conversation(activity, map["actor"])
-      participations = get_participations(conversation)
-      stream_out(activity)
-      stream_out_participations(participations)
       {:ok, activity}
     else
       %Activity{} = activity ->
@@ -205,6 +199,15 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     end
   end
 
+  def notify_and_stream(activity) do
+    Notification.create_notifications(activity)
+
+    conversation = create_or_bump_conversation(activity, activity.actor)
+    participations = get_participations(conversation)
+    stream_out(activity)
+    stream_out_participations(participations)
+  end
+
   defp create_or_bump_conversation(activity, actor) do
     with {:ok, conversation} <- Conversation.create_or_bump_for(activity),
          %User{} = user <- User.get_cached_by_ap_id(actor),
@@ -281,6 +284,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
          _ <- increase_poll_votes_if_vote(create_data),
          {:quick_insert, false, activity} <- {:quick_insert, quick_insert?, activity},
          {:ok, _actor} <- increase_note_count_if_public(actor, activity),
+         _ <- notify_and_stream(activity),
          :ok <- maybe_federate(activity) do
       {:ok, activity}
     else
@@ -308,6 +312,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
              additional
            ),
          {:ok, activity} <- insert(listen_data, local),
+         _ <- notify_and_stream(activity),
          :ok <- maybe_federate(activity) do
       {:ok, activity}
     end
@@ -332,6 +337,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
            %{"to" => to, "type" => type, "actor" => actor.ap_id, "object" => object}
            |> Utils.maybe_put("id", activity_id),
          {:ok, activity} <- insert(data, local),
+         _ <- notify_and_stream(activity),
          :ok <- maybe_federate(activity) do
       {:ok, activity}
     end
@@ -351,6 +357,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
          },
          data <- Utils.maybe_put(data, "id", activity_id),
          {:ok, activity} <- insert(data, local),
+         _ <- notify_and_stream(activity),
          :ok <- maybe_federate(activity) do
       {:ok, activity}
     end
@@ -372,6 +379,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
          reaction_data <- make_emoji_reaction_data(user, object, emoji, activity_id),
          {:ok, activity} <- insert(reaction_data, local),
          {:ok, object} <- add_emoji_reaction_to_object(activity, object),
+         _ <- notify_and_stream(activity),
          :ok <- maybe_federate(activity) do
       {:ok, activity, object}
     else
@@ -398,6 +406,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
          unreact_data <- make_undo_data(user, reaction_activity, activity_id),
          {:ok, activity} <- insert(unreact_data, local),
          {:ok, object} <- remove_emoji_reaction_from_object(reaction_activity, object),
+         _ <- notify_and_stream(activity),
          :ok <- maybe_federate(activity) do
       {:ok, activity, object}
     else
@@ -420,6 +429,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
          {:ok, unlike_activity} <- insert(unlike_data, local),
          {:ok, _activity} <- Repo.delete(like_activity),
          {:ok, object} <- remove_like_from_object(like_activity, object),
+         _ <- notify_and_stream(unlike_activity),
          :ok <- maybe_federate(unlike_activity) do
       {:ok, unlike_activity, like_activity, object}
     else
@@ -449,6 +459,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
          announce_data <- make_announce_data(user, object, activity_id, public),
          {:ok, activity} <- insert(announce_data, local),
          {:ok, object} <- add_announce_to_object(activity, object),
+         _ <- notify_and_stream(activity),
          :ok <- maybe_federate(activity) do
       {:ok, activity, object}
     else
@@ -475,6 +486,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
          unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
          {:ok, unannounce_activity} <- insert(unannounce_data, local),
+         _ <- notify_and_stream(unannounce_activity),
          :ok <- maybe_federate(unannounce_activity),
          {:ok, _activity} <- Repo.delete(announce_activity),
          {:ok, object} <- remove_announce_from_object(announce_activity, object) do
@@ -497,6 +509,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   defp do_follow(follower, followed, activity_id, local) do
     with data <- make_follow_data(follower, followed, activity_id),
          {:ok, activity} <- insert(data, local),
+         _ <- notify_and_stream(activity),
          :ok <- maybe_federate(activity) do
       {:ok, activity}
     else
@@ -518,6 +531,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
          {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
          unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
          {:ok, activity} <- insert(unfollow_data, local),
+         _ <- notify_and_stream(activity),
          :ok <- maybe_federate(activity) do
       {:ok, activity}
     else
@@ -547,6 +561,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     with true <- outgoing_blocks,
          block_data <- make_block_data(blocker, blocked, activity_id),
          {:ok, activity} <- insert(block_data, local),
+         _ <- notify_and_stream(activity),
          :ok <- maybe_federate(activity) do
       {:ok, activity}
     else
@@ -567,6 +582,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
          unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
          {:ok, activity} <- insert(unblock_data, local),
+         _ <- notify_and_stream(activity),
          :ok <- maybe_federate(activity) do
       {:ok, activity}
     else
@@ -601,6 +617,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     with flag_data <- make_flag_data(params, additional),
          {:ok, activity} <- insert(flag_data, local),
          {:ok, stripped_activity} <- strip_report_status_data(activity),
+         _ <- notify_and_stream(activity),
          :ok <- maybe_federate(stripped_activity) do
       User.all_superusers()
       |> Enum.filter(fn user -> not is_nil(user.email) end)
@@ -624,7 +641,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     }
 
     with true <- origin.ap_id in target.also_known_as,
-         {:ok, activity} <- insert(params, local) do
+         {:ok, activity} <- insert(params, local),
+         _ <- notify_and_stream(activity) do
       maybe_federate(activity)
 
       BackgroundWorker.enqueue("move_following", %{

lib/pleroma/web/admin_api/admin_api_controller.ex

@@ -10,6 +10,7 @@ defmodule Pleroma.Web.AdminAPI.AdminAPIController do
   alias Pleroma.Activity
   alias Pleroma.Config
   alias Pleroma.ConfigDB
+  alias Pleroma.MFA
   alias Pleroma.ModerationLog
   alias Pleroma.Plugs.OAuthScopesPlug
   alias Pleroma.ReportNote
@@ -61,6 +62,7 @@ defmodule Pleroma.Web.AdminAPI.AdminAPIController do
            :right_add,
            :right_add_multiple,
            :right_delete,
+           :disable_mfa,
            :right_delete_multiple,
            :update_user_credentials
          ]
@@ -674,6 +676,18 @@ defmodule Pleroma.Web.AdminAPI.AdminAPIController do
     json_response(conn, :no_content, "")
   end
 
+  @doc "Disable mfa for user's account."
+  def disable_mfa(conn, %{"nickname" => nickname}) do
+    case User.get_by_nickname(nickname) do
+      %User{} = user ->
+        MFA.disable(user)
+        json(conn, nickname)
+
+      _ ->
+        {:error, :not_found}
+    end
+  end
+
   @doc "Show a given user's credentials"
   def show_user_credentials(%{assigns: %{user: admin}} = conn, %{"nickname" => nickname}) do
     with %User{} = user <- User.get_cached_by_nickname_or_id(nickname) do

lib/pleroma/web/api_spec/operations/poll_operation.ex

@@ -0,0 +1,76 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.ApiSpec.PollOperation do
+  alias OpenApiSpex.Operation
+  alias OpenApiSpex.Schema
+  alias Pleroma.Web.ApiSpec.Schemas.ApiError
+  alias Pleroma.Web.ApiSpec.Schemas.FlakeID
+  alias Pleroma.Web.ApiSpec.Schemas.Poll
+
+  import Pleroma.Web.ApiSpec.Helpers
+
+  def open_api_operation(action) do
+    operation = String.to_existing_atom("#{action}_operation")
+    apply(__MODULE__, operation, [])
+  end
+
+  def show_operation do
+    %Operation{
+      tags: ["Polls"],
+      summary: "View a poll",
+      security: [%{"oAuth" => ["read:statuses"]}],
+      parameters: [id_param()],
+      operationId: "PollController.show",
+      responses: %{
+        200 => Operation.response("Poll", "application/json", Poll),
+        404 => Operation.response("Error", "application/json", ApiError)
+      }
+    }
+  end
+
+  def vote_operation do
+    %Operation{
+      tags: ["Polls"],
+      summary: "Vote on a poll",
+      parameters: [id_param()],
+      operationId: "PollController.vote",
+      requestBody: vote_request(),
+      security: [%{"oAuth" => ["write:statuses"]}],
+      responses: %{
+        200 => Operation.response("Poll", "application/json", Poll),
+        422 => Operation.response("Error", "application/json", ApiError),
+        404 => Operation.response("Error", "application/json", ApiError)
+      }
+    }
+  end
+
+  defp id_param do
+    Operation.parameter(:id, :path, FlakeID, "Poll ID",
+      example: "123",
+      required: true
+    )
+  end
+
+  defp vote_request do
+    request_body(
+      "Parameters",
+      %Schema{
+        type: :object,
+        properties: %{
+          choices: %Schema{
+            type: :array,
+            items: %Schema{type: :integer},
+            description: "Array of own votes containing index for each option (starting from 0)"
+          }
+        },
+        required: [:choices]
+      },
+      required: true,
+      example: %{
+        "choices" => [0, 1, 2]
+      }
+    )
+  end
+end

lib/pleroma/web/api_spec/schemas/poll.ex

@@ -11,26 +11,72 @@ defmodule Pleroma.Web.ApiSpec.Schemas.Poll do
 
   OpenApiSpex.schema(%{
     title: "Poll",
-    description: "Response schema for account custom fields",
+    description: "Represents a poll attached to a status",
     type: :object,
     properties: %{
       id: FlakeID,
-      expires_at: %Schema{type: :string, format: "date-time"},
-      expired: %Schema{type: :boolean},
-      multiple: %Schema{type: :boolean},
-      votes_count: %Schema{type: :integer},
-      voted: %Schema{type: :boolean},
-      emojis: %Schema{type: :array, items: Emoji},
+      expires_at: %Schema{
+        type: :string,
+        format: :"date-time",
+        nullable: true,
+        description: "When the poll ends"
+      },
+      expired: %Schema{type: :boolean, description: "Is the poll currently expired?"},
+      multiple: %Schema{
+        type: :boolean,
+        description: "Does the poll allow multiple-choice answers?"
+      },
+      votes_count: %Schema{
+        type: :integer,
+        nullable: true,
+        description: "How many votes have been received. Number, or null if `multiple` is false."
+      },
+      voted: %Schema{
+        type: :boolean,
+        nullable: true,
+        description:
+          "When called with a user token, has the authorized user voted? Boolean, or null if no current user."
+      },
+      emojis: %Schema{
+        type: :array,
+        items: Emoji,
+        description: "Custom emoji to be used for rendering poll options."
+      },
       options: %Schema{
         type: :array,
         items: %Schema{
+          title: "PollOption",
           type: :object,
           properties: %{
             title: %Schema{type: :string},
             votes_count: %Schema{type: :integer}
           }
-        }
+        },
+        description: "Possible answers for the poll."
       }
+    },
+    example: %{
+      id: "34830",
+      expires_at: "2019-12-05T04:05:08.302Z",
+      expired: true,
+      multiple: false,
+      votes_count: 10,
+      voters_count: nil,
+      voted: true,
+      own_votes: [
+        1
+      ],
+      options: [
+        %{
+          title: "accept",
+          votes_count: 6
+        },
+        %{
+          title: "deny",
+          votes_count: 4
+        }
+      ],
+      emojis: []
     }
   })
 end

lib/pleroma/web/auth/pleroma_authenticator.ex

@@ -19,8 +19,8 @@ defmodule Pleroma.Web.Auth.PleromaAuthenticator do
          {_, true} <- {:checkpw, AuthenticationPlug.checkpw(password, user.password_hash)} do
       {:ok, user}
     else
-      error ->
-        {:error, error}
+      {:error, _reason} = error -> error
+      error -> {:error, error}
     end
   end
 

lib/pleroma/web/auth/totp_authenticator.ex

@@ -0,0 +1,45 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.Auth.TOTPAuthenticator do
+  alias Comeonin.Pbkdf2
+  alias Pleroma.MFA
+  alias Pleroma.MFA.TOTP
+  alias Pleroma.User
+
+  @doc "Verify code or check backup code."
+  @spec verify(String.t(), User.t()) ::
+          {:ok, :pass} | {:error, :invalid_token | :invalid_secret_and_token}
+  def verify(
+        token,
+        %User{
+          multi_factor_authentication_settings:
+            %{enabled: true, totp: %{secret: secret, confirmed: true}} = _
+        } = _user
+      )
+      when is_binary(token) and byte_size(token) > 0 do
+    TOTP.validate_token(secret, token)
+  end
+
+  def verify(_, _), do: {:error, :invalid_token}
+
+  @spec verify_recovery_code(User.t(), String.t()) ::
+          {:ok, :pass} | {:error, :invalid_token}
+  def verify_recovery_code(
+        %User{multi_factor_authentication_settings: %{enabled: true, backup_codes: codes}} = user,
+        code
+      )
+      when is_list(codes) and is_binary(code) do
+    hash_code = Enum.find(codes, fn hash -> Pbkdf2.checkpw(code, hash) end)
+
+    if hash_code do
+      MFA.invalidate_backup_code(user, hash_code)
+      {:ok, :pass}
+    else
+      {:error, :invalid_token}
+    end
+  end
+
+  def verify_recovery_code(_, _), do: {:error, :invalid_token}
+end

lib/pleroma/web/common_api/utils.ex

@@ -402,6 +402,7 @@ defmodule Pleroma.Web.CommonAPI.Utils do
     end
   end
 
+  @spec confirm_current_password(User.t(), String.t()) :: {:ok, User.t()} | {:error, String.t()}
   def confirm_current_password(user, password) do
     with %User{local: true} = db_user <- User.get_cached_by_id(user.id),
          true <- AuthenticationPlug.checkpw(password, db_user.password_hash) do

lib/pleroma/web/mastodon_api/controllers/poll_controller.ex

@@ -15,6 +15,8 @@ defmodule Pleroma.Web.MastodonAPI.PollController do
 
   action_fallback(Pleroma.Web.MastodonAPI.FallbackController)
 
+  plug(Pleroma.Web.ApiSpec.CastAndValidate)
+
   plug(
     OAuthScopesPlug,
     %{scopes: ["read:statuses"], fallback: :proceed_unauthenticated} when action == :show
@@ -22,8 +24,10 @@ defmodule Pleroma.Web.MastodonAPI.PollController do
 
   plug(OAuthScopesPlug, %{scopes: ["write:statuses"]} when action == :vote)
 
+  defdelegate open_api_operation(action), to: Pleroma.Web.ApiSpec.PollOperation
+
   @doc "GET /api/v1/polls/:id"
-  def show(%{assigns: %{user: user}} = conn, %{"id" => id}) do
+  def show(%{assigns: %{user: user}} = conn, %{id: id}) do
     with %Object{} = object <- Object.get_by_id_and_maybe_refetch(id, interval: 60),
          %Activity{} = activity <- Activity.get_create_by_object_ap_id(object.data["id"]),
          true <- Visibility.visible_for_user?(activity, user) do
@@ -35,7 +39,7 @@ defmodule Pleroma.Web.MastodonAPI.PollController do
   end
 
   @doc "POST /api/v1/polls/:id/votes"
-  def vote(%{assigns: %{user: user}} = conn, %{"id" => id, "choices" => choices}) do
+  def vote(%{assigns: %{user: user}, body_params: %{choices: choices}} = conn, %{id: id}) do
     with %Object{data: %{"type" => "Question"}} = object <- Object.get_by_id(id),
          %Activity{} = activity <- Activity.get_create_by_object_ap_id(object.data["id"]),
          true <- Visibility.visible_for_user?(activity, user),

lib/pleroma/web/mastodon_api/websocket_handler.ex

@@ -12,6 +12,11 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
 
   @behaviour :cowboy_websocket
 
+  # Cowboy timeout period.
+  @timeout :timer.seconds(30)
+  # Hibernate every X messages
+  @hibernate_every 100
+
   @streams [
     "public",
     "public:local",
@@ -25,9 +30,6 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
   ]
   @anonymous_streams ["public", "public:local", "hashtag"]
 
-  # Handled by periodic keepalive in Pleroma.Web.Streamer.Ping.
-  @timeout :infinity
-
   def init(%{qs: qs} = req, state) do
     with params <- :cow_qs.parse_qs(qs),
          sec_websocket <- :cowboy_req.header("sec-websocket-protocol", req, nil),
@@ -42,7 +44,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
           req
         end
 
-      {:cowboy_websocket, req, %{user: user, topic: topic}, %{idle_timeout: @timeout}}
+      {:cowboy_websocket, req, %{user: user, topic: topic, count: 0}, %{idle_timeout: @timeout}}
     else
       {:error, code} ->
         Logger.debug("#{__MODULE__} denied connection: #{inspect(code)} - #{inspect(req)}")
@@ -57,7 +59,13 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
   end
 
   def websocket_init(state) do
-    send(self(), :subscribe)
+    Logger.debug(
+      "#{__MODULE__} accepted websocket connection for user #{
+        (state.user || %{id: "anonymous"}).id
+      }, topic #{state.topic}"
+    )
+
+    Streamer.add_socket(state.topic, state.user)
     {:ok, state}
   end
 
@@ -66,19 +74,24 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
     {:ok, state}
   end
 
-  def websocket_info(:subscribe, state) do
-    Logger.debug(
-      "#{__MODULE__} accepted websocket connection for user #{
-        (state.user || %{id: "anonymous"}).id
-      }, topic #{state.topic}"
-    )
+  def websocket_info({:render_with_user, view, template, item}, state) do
+    user = %User{} = User.get_cached_by_ap_id(state.user.ap_id)
 
-    Streamer.add_socket(state.topic, streamer_socket(state))
-    {:ok, state}
+    unless Streamer.filtered_by_user?(user, item) do
+      websocket_info({:text, view.render(template, user, item)}, %{state | user: user})
+    else
+      {:ok, state}
+    end
   end
 
   def websocket_info({:text, message}, state) do
-    {:reply, {:text, message}, state}
+    # If the websocket processed X messages, force an hibernate/GC.
+    # We don't hibernate at every message to balance CPU usage/latency with RAM usage.
+    if state.count > @hibernate_every do
+      {:reply, {:text, message}, %{state | count: 0}, :hibernate}
+    else
+      {:reply, {:text, message}, %{state | count: state.count + 1}}
+    end
   end
 
   def terminate(reason, _req, state) do
@@ -88,7 +101,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
       }, topic #{state.topic || "?"}: #{inspect(reason)}"
     )
 
-    Streamer.remove_socket(state.topic, streamer_socket(state))
+    Streamer.remove_socket(state.topic)
     :ok
   end
 
@@ -136,8 +149,4 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
   end
 
   defp expand_topic(topic, _), do: topic
-
-  defp streamer_socket(state) do
-    %{transport_pid: self(), assigns: state}
-  end
 end

lib/pleroma/web/oauth/mfa_controller.ex

@@ -0,0 +1,97 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.OAuth.MFAController do
+  @moduledoc """
+  The model represents api to use Multi Factor authentications.
+  """
+
+  use Pleroma.Web, :controller
+
+  alias Pleroma.MFA
+  alias Pleroma.Web.Auth.TOTPAuthenticator
+  alias Pleroma.Web.OAuth.MFAView, as: View
+  alias Pleroma.Web.OAuth.OAuthController
+  alias Pleroma.Web.OAuth.Token
+
+  plug(:fetch_session when action in [:show, :verify])
+  plug(:fetch_flash when action in [:show, :verify])
+
+  @doc """
+  Display form to input mfa code or recovery code.
+  """
+  def show(conn, %{"mfa_token" => mfa_token} = params) do
+    template = Map.get(params, "challenge_type", "totp")
+
+    conn
+    |> put_view(View)
+    |> render("#{template}.html", %{
+      mfa_token: mfa_token,
+      redirect_uri: params["redirect_uri"],
+      state: params["state"]
+    })
+  end
+
+  @doc """
+  Verification code and continue authorization.
+  """
+  def verify(conn, %{"mfa" => %{"mfa_token" => mfa_token} = mfa_params} = _) do
+    with {:ok, %{user: user, authorization: auth}} <- MFA.Token.validate(mfa_token),
+         {:ok, _} <- validates_challenge(user, mfa_params) do
+      conn
+      |> OAuthController.after_create_authorization(auth, %{
+        "authorization" => %{
+          "redirect_uri" => mfa_params["redirect_uri"],
+          "state" => mfa_params["state"]
+        }
+      })
+    else
+      _ ->
+        conn
+        |> put_flash(:error, "Two-factor authentication failed.")
+        |> put_status(:unauthorized)
+        |> show(mfa_params)
+    end
+  end
+
+  @doc """
+  Verification second step of MFA (or recovery) and returns access token.
+
+  ## Endpoint
+  POST /oauth/mfa/challenge
+
+  params:
+  `client_id`
+  `client_secret`
+  `mfa_token` - access token to check second step of mfa
+  `challenge_type` - 'totp' or 'recovery'
+  `code`
+
+  """
+  def challenge(conn, %{"mfa_token" => mfa_token} = params) do
+    with {:ok, app} <- Token.Utils.fetch_app(conn),
+         {:ok, %{user: user, authorization: auth}} <- MFA.Token.validate(mfa_token),
+         {:ok, _} <- validates_challenge(user, params),
+         {:ok, token} <- Token.exchange_token(app, auth) do
+      json(conn, Token.Response.build(user, token))
+    else
+      _error ->
+        conn
+        |> put_status(400)
+        |> json(%{error: "Invalid code"})
+    end
+  end
+
+  # Verify TOTP Code
+  defp validates_challenge(user, %{"challenge_type" => "totp", "code" => code} = _) do
+    TOTPAuthenticator.verify(code, user)
+  end
+
+  # Verify Recovery Code
+  defp validates_challenge(user, %{"challenge_type" => "recovery", "code" => code} = _) do
+    TOTPAuthenticator.verify_recovery_code(user, code)
+  end
+
+  defp validates_challenge(_, _), do: {:error, :unsupported_challenge_type}
+end

lib/pleroma/web/oauth/mfa_view.ex

@@ -0,0 +1,8 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.OAuth.MFAView do
+  use Pleroma.Web, :view
+  import Phoenix.HTML.Form
+end

lib/pleroma/web/oauth/oauth_controller.ex

@@ -6,6 +6,7 @@ defmodule Pleroma.Web.OAuth.OAuthController do
   use Pleroma.Web, :controller
 
   alias Pleroma.Helpers.UriHelper
+  alias Pleroma.MFA
   alias Pleroma.Plugs.RateLimiter
   alias Pleroma.Registration
   alias Pleroma.Repo
@@ -14,6 +15,7 @@ defmodule Pleroma.Web.OAuth.OAuthController do
   alias Pleroma.Web.ControllerHelper
   alias Pleroma.Web.OAuth.App
   alias Pleroma.Web.OAuth.Authorization
+  alias Pleroma.Web.OAuth.MFAController
   alias Pleroma.Web.OAuth.Scopes
   alias Pleroma.Web.OAuth.Token
   alias Pleroma.Web.OAuth.Token.Strategy.RefreshToken
@@ -121,7 +123,8 @@ defmodule Pleroma.Web.OAuth.OAuthController do
         %{"authorization" => _} = params,
         opts \\ []
       ) do
-    with {:ok, auth} <- do_create_authorization(conn, params, opts[:user]) do
+    with {:ok, auth, user} <- do_create_authorization(conn, params, opts[:user]),
+         {:mfa_required, _, _, false} <- {:mfa_required, user, auth, MFA.require?(user)} do
       after_create_authorization(conn, auth, params)
     else
       error ->
@@ -179,6 +182,22 @@ defmodule Pleroma.Web.OAuth.OAuthController do
     |> authorize(params)
   end
 
+  defp handle_create_authorization_error(
+         %Plug.Conn{} = conn,
+         {:mfa_required, user, auth, _},
+         params
+       ) do
+    {:ok, token} = MFA.Token.create_token(user, auth)
+
+    data = %{
+      "mfa_token" => token.token,
+      "redirect_uri" => params["authorization"]["redirect_uri"],
+      "state" => params["authorization"]["state"]
+    }
+
+    MFAController.show(conn, data)
+  end
+
   defp handle_create_authorization_error(
          %Plug.Conn{} = conn,
          {:account_status, :password_reset_pending},
@@ -231,7 +250,8 @@ defmodule Pleroma.Web.OAuth.OAuthController do
 
       json(conn, Token.Response.build(user, token, response_attrs))
     else
-      _error -> render_invalid_credentials_error(conn)
+      error ->
+        handle_token_exchange_error(conn, error)
     end
   end
 
@@ -244,6 +264,7 @@ defmodule Pleroma.Web.OAuth.OAuthController do
          {:account_status, :active} <- {:account_status, User.account_status(user)},
          {:ok, scopes} <- validate_scopes(app, params),
          {:ok, auth} <- Authorization.create_authorization(app, user, scopes),
+         {:mfa_required, _, _, false} <- {:mfa_required, user, auth, MFA.require?(user)},
          {:ok, token} <- Token.exchange_token(app, auth) do
       json(conn, Token.Response.build(user, token))
     else
@@ -270,13 +291,20 @@ defmodule Pleroma.Web.OAuth.OAuthController do
          {:ok, token} <- Token.exchange_token(app, auth) do
       json(conn, Token.Response.build_for_client_credentials(token))
     else
-      _error -> render_invalid_credentials_error(conn)
+      _error ->
+        handle_token_exchange_error(conn, :invalid_credentails)
     end
   end
 
   # Bad request
   def token_exchange(%Plug.Conn{} = conn, params), do: bad_request(conn, params)
 
+  defp handle_token_exchange_error(%Plug.Conn{} = conn, {:mfa_required, user, auth, _}) do
+    conn
+    |> put_status(:forbidden)
+    |> json(build_and_response_mfa_token(user, auth))
+  end
+
   defp handle_token_exchange_error(%Plug.Conn{} = conn, {:account_status, :deactivated}) do
     render_error(
       conn,
@@ -434,7 +462,8 @@ defmodule Pleroma.Web.OAuth.OAuthController do
   def register(%Plug.Conn{} = conn, %{"authorization" => _, "op" => "connect"} = params) do
     with registration_id when not is_nil(registration_id) <- get_session_registration_id(conn),
          %Registration{} = registration <- Repo.get(Registration, registration_id),
-         {_, {:ok, auth}} <- {:create_authorization, do_create_authorization(conn, params)},
+         {_, {:ok, auth, _user}} <-
+           {:create_authorization, do_create_authorization(conn, params)},
          %User{} = user <- Repo.preload(auth, :user).user,
          {:ok, _updated_registration} <- Registration.bind_to_user(registration, user) do
       conn
@@ -500,8 +529,9 @@ defmodule Pleroma.Web.OAuth.OAuthController do
          %App{} = app <- Repo.get_by(App, client_id: client_id),
          true <- redirect_uri in String.split(app.redirect_uris),
          {:ok, scopes} <- validate_scopes(app, auth_attrs),
-         {:account_status, :active} <- {:account_status, User.account_status(user)} do
-      Authorization.create_authorization(app, user, scopes)
+         {:account_status, :active} <- {:account_status, User.account_status(user)},
+         {:ok, auth} <- Authorization.create_authorization(app, user, scopes) do
+      {:ok, auth, user}
     end
   end
 
@@ -515,6 +545,12 @@ defmodule Pleroma.Web.OAuth.OAuthController do
   defp put_session_registration_id(%Plug.Conn{} = conn, registration_id),
     do: put_session(conn, :registration_id, registration_id)
 
+  defp build_and_response_mfa_token(user, auth) do
+    with {:ok, token} <- MFA.Token.create_token(user, auth) do
+      Token.Response.build_for_mfa_token(user, token)
+    end
+  end
+
   @spec validate_scopes(App.t(), map()) ::
           {:ok, list()} | {:error, :missing_scopes | :unsupported_scopes}
   defp validate_scopes(%App{} = app, params) do

lib/pleroma/web/oauth/token/clean_worker.ex

@@ -0,0 +1,38 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.OAuth.Token.CleanWorker do
+  @moduledoc """
+  The module represents functions to clean an expired OAuth and MFA tokens.
+  """
+  use GenServer
+
+  @ten_seconds 10_000
+  @one_day 86_400_000
+
+  alias Pleroma.MFA
+  alias Pleroma.Web.OAuth
+  alias Pleroma.Workers.BackgroundWorker
+
+  def start_link(_), do: GenServer.start_link(__MODULE__, %{})
+
+  def init(_) do
+    Process.send_after(self(), :perform, @ten_seconds)
+    {:ok, nil}
+  end
+
+  @doc false
+  def handle_info(:perform, state) do
+    BackgroundWorker.enqueue("clean_expired_tokens", %{})
+    interval = Pleroma.Config.get([:oauth2, :clean_expired_tokens_interval], @one_day)
+
+    Process.send_after(self(), :perform, interval)
+    {:noreply, state}
+  end
+
+  def perform(:clean) do
+    OAuth.Token.delete_expired_tokens()
+    MFA.Token.delete_expired_tokens()
+  end
+end

lib/pleroma/web/oauth/token/response.ex

@@ -5,6 +5,7 @@
 defmodule Pleroma.Web.OAuth.Token.Response do
   @moduledoc false
 
+  alias Pleroma.MFA
   alias Pleroma.User
   alias Pleroma.Web.OAuth.Token.Utils
 
@@ -32,5 +33,13 @@ defmodule Pleroma.Web.OAuth.Token.Response do
     }
   end
 
+  def build_for_mfa_token(user, mfa_token) do
+    %{
+      error: "mfa_required",
+      mfa_token: mfa_token.token,
+      supported_challenge_types: MFA.supported_methods(user)
+    }
+  end
+
   defp expires_in, do: Pleroma.Config.get([:oauth2, :token_expires_in], 600)
 end

lib/pleroma/web/pleroma_api/controllers/two_factor_authentication_controller.ex

@@ -0,0 +1,133 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.PleromaAPI.TwoFactorAuthenticationController do
+  @moduledoc "The module represents actions to manage MFA"
+  use Pleroma.Web, :controller
+
+  import Pleroma.Web.ControllerHelper, only: [json_response: 3]
+
+  alias Pleroma.MFA
+  alias Pleroma.MFA.TOTP
+  alias Pleroma.Plugs.OAuthScopesPlug
+  alias Pleroma.Web.CommonAPI.Utils
+
+  plug(OAuthScopesPlug, %{scopes: ["read:security"]} when action in [:settings])
+
+  plug(
+    OAuthScopesPlug,
+    %{scopes: ["write:security"]} when action in [:setup, :confirm, :disable, :backup_codes]
+  )
+
+  @doc """
+  Gets user multi factor authentication settings
+
+  ## Endpoint
+  GET /api/pleroma/accounts/mfa
+
+  """
+  def settings(%{assigns: %{user: user}} = conn, _params) do
+    json(conn, %{settings: MFA.mfa_settings(user)})
+  end
+
+  @doc """
+  Prepare setup mfa method
+
+  ## Endpoint
+  GET /api/pleroma/accounts/mfa/setup/[:method]
+
+  """
+  def setup(%{assigns: %{user: user}} = conn, %{"method" => "totp"} = _params) do
+    with {:ok, user} <- MFA.setup_totp(user),
+         %{secret: secret} = _ <- user.multi_factor_authentication_settings.totp do
+      provisioning_uri = TOTP.provisioning_uri(secret, "#{user.email}")
+
+      json(conn, %{provisioning_uri: provisioning_uri, key: secret})
+    else
+      {:error, message} ->
+        json_response(conn, :unprocessable_entity, %{error: message})
+    end
+  end
+
+  def setup(conn, _params) do
+    json_response(conn, :bad_request, %{error: "undefined method"})
+  end
+
+  @doc """
+  Confirms setup and enable mfa method
+
+  ## Endpoint
+  POST /api/pleroma/accounts/mfa/confirm/:method
+
+  - params:
+  `code` - confirmation code
+  `password` - current password
+  """
+  def confirm(
+        %{assigns: %{user: user}} = conn,
+        %{"method" => "totp", "password" => _, "code" => _} = params
+      ) do
+    with {:ok, _user} <- Utils.confirm_current_password(user, params["password"]),
+         {:ok, _user} <- MFA.confirm_totp(user, params) do
+      json(conn, %{})
+    else
+      {:error, message} ->
+        json_response(conn, :unprocessable_entity, %{error: message})
+    end
+  end
+
+  def confirm(conn, _) do
+    json_response(conn, :bad_request, %{error: "undefined mfa method"})
+  end
+
+  @doc """
+  Disable mfa method and disable mfa if need.
+  """
+  def disable(%{assigns: %{user: user}} = conn, %{"method" => "totp"} = params) do
+    with {:ok, user} <- Utils.confirm_current_password(user, params["password"]),
+         {:ok, _user} <- MFA.disable_totp(user) do
+      json(conn, %{})
+    else
+      {:error, message} ->
+        json_response(conn, :unprocessable_entity, %{error: message})
+    end
+  end
+
+  def disable(%{assigns: %{user: user}} = conn, %{"method" => "mfa"} = params) do
+    with {:ok, user} <- Utils.confirm_current_password(user, params["password"]),
+         {:ok, _user} <- MFA.disable(user) do
+      json(conn, %{})
+    else
+      {:error, message} ->
+        json_response(conn, :unprocessable_entity, %{error: message})
+    end
+  end
+
+  def disable(conn, _) do
+    json_response(conn, :bad_request, %{error: "undefined mfa method"})
+  end
+
+  @doc """
+  Generates backup codes.
+
+  ## Endpoint
+  GET /api/pleroma/accounts/mfa/backup_codes
+
+  ## Response
+  ### Success
+  `{codes: [codes]}`
+
+  ### Error
+  `{error: [error_message]}`
+
+  """
+  def backup_codes(%{assigns: %{user: user}} = conn, _params) do
+    with {:ok, codes} <- MFA.generate_backup_codes(user) do
+      json(conn, %{codes: codes})
+    else
+      {:error, message} ->
+        json_response(conn, :unprocessable_entity, %{error: message})
+    end
+  end
+end

lib/pleroma/web/router.ex

@@ -132,6 +132,7 @@ defmodule Pleroma.Web.Router do
     post("/users/follow", AdminAPIController, :user_follow)
     post("/users/unfollow", AdminAPIController, :user_unfollow)
 
+    put("/users/disable_mfa", AdminAPIController, :disable_mfa)
     delete("/users", AdminAPIController, :user_delete)
     post("/users", AdminAPIController, :users_create)
     patch("/users/:nickname/toggle_activation", AdminAPIController, :user_toggle_activation)
@@ -258,6 +259,16 @@ defmodule Pleroma.Web.Router do
     post("/follow_import", UtilController, :follow_import)
   end
 
+  scope "/api/pleroma", Pleroma.Web.PleromaAPI do
+    pipe_through(:authenticated_api)
+
+    get("/accounts/mfa", TwoFactorAuthenticationController, :settings)
+    get("/accounts/mfa/backup_codes", TwoFactorAuthenticationController, :backup_codes)
+    get("/accounts/mfa/setup/:method", TwoFactorAuthenticationController, :setup)
+    post("/accounts/mfa/confirm/:method", TwoFactorAuthenticationController, :confirm)
+    delete("/accounts/mfa/:method", TwoFactorAuthenticationController, :disable)
+  end
+
   scope "/oauth", Pleroma.Web.OAuth do
     scope [] do
       pipe_through(:oauth)
@@ -269,6 +280,10 @@ defmodule Pleroma.Web.Router do
     post("/revoke", OAuthController, :token_revoke)
     get("/registration_details", OAuthController, :registration_details)
 
+    post("/mfa/challenge", MFAController, :challenge)
+    post("/mfa/verify", MFAController, :verify, as: :mfa_verify)
+    get("/mfa", MFAController, :show)
+
     scope [] do
       pipe_through(:browser)
 

lib/pleroma/web/streamer/ping.ex

@@ -1,37 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Web.Streamer.Ping do
-  use GenServer
-  require Logger
-
-  alias Pleroma.Web.Streamer.State
-  alias Pleroma.Web.Streamer.StreamerSocket
-
-  @keepalive_interval :timer.seconds(30)
-
-  def start_link(opts) do
-    ping_interval = Keyword.get(opts, :ping_interval, @keepalive_interval)
-    GenServer.start_link(__MODULE__, %{ping_interval: ping_interval}, name: __MODULE__)
-  end
-
-  def init(%{ping_interval: ping_interval} = args) do
-    Process.send_after(self(), :ping, ping_interval)
-    {:ok, args}
-  end
-
-  def handle_info(:ping, %{ping_interval: ping_interval} = state) do
-    State.get_sockets()
-    |> Map.values()
-    |> List.flatten()
-    |> Enum.each(fn %StreamerSocket{transport_pid: transport_pid} ->
-      Logger.debug("Sending keepalive ping")
-      send(transport_pid, {:text, ""})
-    end)
-
-    Process.send_after(self(), :ping, ping_interval)
-
-    {:noreply, state}
-  end
-end

lib/pleroma/web/streamer/state.ex

@@ -1,82 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Web.Streamer.State do
-  use GenServer
-  require Logger
-
-  alias Pleroma.Web.Streamer.StreamerSocket
-
-  @env Mix.env()
-
-  def start_link(_) do
-    GenServer.start_link(__MODULE__, %{sockets: %{}}, name: __MODULE__)
-  end
-
-  def add_socket(topic, socket) do
-    GenServer.call(__MODULE__, {:add, topic, socket})
-  end
-
-  def remove_socket(topic, socket) do
-    do_remove_socket(@env, topic, socket)
-  end
-
-  def get_sockets do
-    %{sockets: stream_sockets} = GenServer.call(__MODULE__, :get_state)
-    stream_sockets
-  end
-
-  def init(init_arg) do
-    {:ok, init_arg}
-  end
-
-  def handle_call(:get_state, _from, state) do
-    {:reply, state, state}
-  end
-
-  def handle_call({:add, topic, socket}, _from, %{sockets: sockets} = state) do
-    internal_topic = internal_topic(topic, socket)
-    stream_socket = StreamerSocket.from_socket(socket)
-
-    sockets_for_topic =
-      sockets
-      |> Map.get(internal_topic, [])
-      |> List.insert_at(0, stream_socket)
-      |> Enum.uniq()
-
-    state = put_in(state, [:sockets, internal_topic], sockets_for_topic)
-    Logger.debug("Got new conn for #{topic}")
-    {:reply, state, state}
-  end
-
-  def handle_call({:remove, topic, socket}, _from, %{sockets: sockets} = state) do
-    internal_topic = internal_topic(topic, socket)
-    stream_socket = StreamerSocket.from_socket(socket)
-
-    sockets_for_topic =
-      sockets
-      |> Map.get(internal_topic, [])
-      |> List.delete(stream_socket)
-
-    state = Kernel.put_in(state, [:sockets, internal_topic], sockets_for_topic)
-    {:reply, state, state}
-  end
-
-  defp do_remove_socket(:test, _, _) do
-    :ok
-  end
-
-  defp do_remove_socket(_env, topic, socket) do
-    GenServer.call(__MODULE__, {:remove, topic, socket})
-  end
-
-  defp internal_topic(topic, socket)
-       when topic in ~w[user user:notification direct] do
-    "#{topic}:#{socket.assigns[:user].id}"
-  end
-
-  defp internal_topic(topic, _) do
-    topic
-  end
-end

lib/pleroma/web/streamer/streamer.ex

@@ -3,53 +3,241 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.Streamer do
-  alias Pleroma.Web.Streamer.State
-  alias Pleroma.Web.Streamer.Worker
+  require Logger
+
+  alias Pleroma.Activity
+  alias Pleroma.Config
+  alias Pleroma.Conversation.Participation
+  alias Pleroma.Notification
+  alias Pleroma.Object
+  alias Pleroma.User
+  alias Pleroma.Web.ActivityPub.ActivityPub
+  alias Pleroma.Web.ActivityPub.Visibility
+  alias Pleroma.Web.CommonAPI
+  alias Pleroma.Web.StreamerView
 
-  @timeout 60_000
   @mix_env Mix.env()
+  @registry Pleroma.Web.StreamerRegistry
 
-  def add_socket(topic, socket) do
-    State.add_socket(topic, socket)
+  def registry, do: @registry
+
+  def add_socket(topic, %User{} = user) do
+    if should_env_send?(), do: Registry.register(@registry, user_topic(topic, user), true)
   end
 
-  def remove_socket(topic, socket) do
-    State.remove_socket(topic, socket)
+  def add_socket(topic, _) do
+    if should_env_send?(), do: Registry.register(@registry, topic, false)
   end
 
-  def get_sockets do
-    State.get_sockets()
+  def remove_socket(topic) do
+    if should_env_send?(), do: Registry.unregister(@registry, topic)
   end
 
-  def stream(topics, items) do
-    if should_send?() do
-      Task.async(fn ->
-        :poolboy.transaction(
-          :streamer_worker,
-          &Worker.stream(&1, topics, items),
-          @timeout
-        )
+  def stream(topics, item) when is_list(topics) do
+    if should_env_send?() do
+      Enum.each(topics, fn t ->
+        spawn(fn -> do_stream(t, item) end)
       end)
     end
+
+    :ok
   end
 
-  def supervisor, do: Pleroma.Web.Streamer.Supervisor
+  def stream(topic, items) when is_list(items) do
+    if should_env_send?() do
+      Enum.each(items, fn i ->
+        spawn(fn -> do_stream(topic, i) end)
+      end)
 
-  defp should_send? do
-    handle_should_send(@mix_env)
-  end
-
-  defp handle_should_send(:test) do
-    case Process.whereis(:streamer_worker) do
-      nil ->
-        false
-
-      pid ->
-        Process.alive?(pid)
+      :ok
     end
   end
 
-  defp handle_should_send(:benchmark), do: false
+  def stream(topic, item) do
+    if should_env_send?() do
+      spawn(fn -> do_stream(topic, item) end)
+    end
 
-  defp handle_should_send(_), do: true
+    :ok
+  end
+
+  def filtered_by_user?(%User{} = user, %Activity{} = item) do
+    %{block: blocked_ap_ids, mute: muted_ap_ids, reblog_mute: reblog_muted_ap_ids} =
+      User.outgoing_relationships_ap_ids(user, [:block, :mute, :reblog_mute])
+
+    recipient_blocks = MapSet.new(blocked_ap_ids ++ muted_ap_ids)
+    recipients = MapSet.new(item.recipients)
+    domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks)
+
+    with parent <- Object.normalize(item) || item,
+         true <-
+           Enum.all?([blocked_ap_ids, muted_ap_ids], &(item.actor not in &1)),
+         true <- item.data["type"] != "Announce" || item.actor not in reblog_muted_ap_ids,
+         true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(parent.data["actor"] not in &1)),
+         true <- MapSet.disjoint?(recipients, recipient_blocks),
+         %{host: item_host} <- URI.parse(item.actor),
+         %{host: parent_host} <- URI.parse(parent.data["actor"]),
+         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
+         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
+         true <- thread_containment(item, user),
+         false <- CommonAPI.thread_muted?(user, item) do
+      false
+    else
+      _ -> true
+    end
+  end
+
+  def filtered_by_user?(%User{} = user, %Notification{activity: activity}) do
+    filtered_by_user?(user, activity)
+  end
+
+  defp do_stream("direct", item) do
+    recipient_topics =
+      User.get_recipients_from_activity(item)
+      |> Enum.map(fn %{id: id} -> "direct:#{id}" end)
+
+    Enum.each(recipient_topics, fn user_topic ->
+      Logger.debug("Trying to push direct message to #{user_topic}\n\n")
+      push_to_socket(user_topic, item)
+    end)
+  end
+
+  defp do_stream("participation", participation) do
+    user_topic = "direct:#{participation.user_id}"
+    Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")
+
+    push_to_socket(user_topic, participation)
+  end
+
+  defp do_stream("list", item) do
+    # filter the recipient list if the activity is not public, see #270.
+    recipient_lists =
+      case Visibility.is_public?(item) do
+        true ->
+          Pleroma.List.get_lists_from_activity(item)
+
+        _ ->
+          Pleroma.List.get_lists_from_activity(item)
+          |> Enum.filter(fn list ->
+            owner = User.get_cached_by_id(list.user_id)
+
+            Visibility.visible_for_user?(item, owner)
+          end)
+      end
+
+    recipient_topics =
+      recipient_lists
+      |> Enum.map(fn %{id: id} -> "list:#{id}" end)
+
+    Enum.each(recipient_topics, fn list_topic ->
+      Logger.debug("Trying to push message to #{list_topic}\n\n")
+      push_to_socket(list_topic, item)
+    end)
+  end
+
+  defp do_stream(topic, %Notification{} = item)
+       when topic in ["user", "user:notification"] do
+    Registry.dispatch(@registry, "#{topic}:#{item.user_id}", fn list ->
+      Enum.each(list, fn {pid, _auth} ->
+        send(pid, {:render_with_user, StreamerView, "notification.json", item})
+      end)
+    end)
+  end
+
+  defp do_stream("user", item) do
+    Logger.debug("Trying to push to users")
+
+    recipient_topics =
+      User.get_recipients_from_activity(item)
+      |> Enum.map(fn %{id: id} -> "user:#{id}" end)
+
+    Enum.each(recipient_topics, fn topic ->
+      push_to_socket(topic, item)
+    end)
+  end
+
+  defp do_stream(topic, item) do
+    Logger.debug("Trying to push to #{topic}")
+    Logger.debug("Pushing item to #{topic}")
+    push_to_socket(topic, item)
+  end
+
+  defp push_to_socket(topic, %Participation{} = participation) do
+    rendered = StreamerView.render("conversation.json", participation)
+
+    Registry.dispatch(@registry, topic, fn list ->
+      Enum.each(list, fn {pid, _} ->
+        send(pid, {:text, rendered})
+      end)
+    end)
+  end
+
+  defp push_to_socket(topic, %Activity{
+         data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
+       }) do
+    rendered = Jason.encode!(%{event: "delete", payload: to_string(deleted_activity_id)})
+
+    Registry.dispatch(@registry, topic, fn list ->
+      Enum.each(list, fn {pid, _} ->
+        send(pid, {:text, rendered})
+      end)
+    end)
+  end
+
+  defp push_to_socket(_topic, %Activity{data: %{"type" => "Delete"}}), do: :noop
+
+  defp push_to_socket(topic, item) do
+    anon_render = StreamerView.render("update.json", item)
+
+    Registry.dispatch(@registry, topic, fn list ->
+      Enum.each(list, fn {pid, auth?} ->
+        if auth? do
+          send(pid, {:render_with_user, StreamerView, "update.json", item})
+        else
+          send(pid, {:text, anon_render})
+        end
+      end)
+    end)
+  end
+
+  defp thread_containment(_activity, %User{skip_thread_containment: true}), do: true
+
+  defp thread_containment(activity, user) do
+    if Config.get([:instance, :skip_thread_containment]) do
+      true
+    else
+      ActivityPub.contain_activity(activity, user)
+    end
+  end
+
+  # In test environement, only return true if the registry is started.
+  # In benchmark environment, returns false.
+  # In any other environment, always returns true.
+  cond do
+    @mix_env == :test ->
+      def should_env_send? do
+        case Process.whereis(@registry) do
+          nil ->
+            false
+
+          pid ->
+            Process.alive?(pid)
+        end
+      end
+
+    @mix_env == :benchmark ->
+      def should_env_send?, do: false
+
+    true ->
+      def should_env_send?, do: true
+  end
+
+  defp user_topic(topic, user)
+       when topic in ~w[user user:notification direct] do
+    "#{topic}:#{user.id}"
+  end
+
+  defp user_topic(topic, _) do
+    topic
+  end
 end

lib/pleroma/web/streamer/streamer_socket.ex

@@ -1,35 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Web.Streamer.StreamerSocket do
-  defstruct transport_pid: nil, user: nil
-
-  alias Pleroma.User
-  alias Pleroma.Web.Streamer.StreamerSocket
-
-  def from_socket(%{
-        transport_pid: transport_pid,
-        assigns: %{user: nil}
-      }) do
-    %StreamerSocket{
-      transport_pid: transport_pid
-    }
-  end
-
-  def from_socket(%{
-        transport_pid: transport_pid,
-        assigns: %{user: %User{} = user}
-      }) do
-    %StreamerSocket{
-      transport_pid: transport_pid,
-      user: user
-    }
-  end
-
-  def from_socket(%{transport_pid: transport_pid}) do
-    %StreamerSocket{
-      transport_pid: transport_pid
-    }
-  end
-end

lib/pleroma/web/streamer/supervisor.ex

@@ -1,37 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Web.Streamer.Supervisor do
-  use Supervisor
-
-  def start_link(opts) do
-    Supervisor.start_link(__MODULE__, opts, name: __MODULE__)
-  end
-
-  def init(args) do
-    children = [
-      {Pleroma.Web.Streamer.State, args},
-      {Pleroma.Web.Streamer.Ping, args},
-      :poolboy.child_spec(:streamer_worker, poolboy_config())
-    ]
-
-    opts = [strategy: :one_for_one, name: Pleroma.Web.Streamer.Supervisor]
-    Supervisor.init(children, opts)
-  end
-
-  defp poolboy_config do
-    opts =
-      Pleroma.Config.get(:streamer,
-        workers: 3,
-        overflow_workers: 2
-      )
-
-    [
-      {:name, {:local, :streamer_worker}},
-      {:worker_module, Pleroma.Web.Streamer.Worker},
-      {:size, opts[:workers]},
-      {:max_overflow, opts[:overflow_workers]}
-    ]
-  end
-end

lib/pleroma/web/streamer/worker.ex

@@ -1,208 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Web.Streamer.Worker do
-  use GenServer
-
-  require Logger
-
-  alias Pleroma.Activity
-  alias Pleroma.Config
-  alias Pleroma.Conversation.Participation
-  alias Pleroma.Notification
-  alias Pleroma.Object
-  alias Pleroma.User
-  alias Pleroma.Web.ActivityPub.ActivityPub
-  alias Pleroma.Web.ActivityPub.Visibility
-  alias Pleroma.Web.CommonAPI
-  alias Pleroma.Web.Streamer.State
-  alias Pleroma.Web.Streamer.StreamerSocket
-  alias Pleroma.Web.StreamerView
-
-  def start_link(_) do
-    GenServer.start_link(__MODULE__, %{}, [])
-  end
-
-  def init(init_arg) do
-    {:ok, init_arg}
-  end
-
-  def stream(pid, topics, items) do
-    GenServer.call(pid, {:stream, topics, items})
-  end
-
-  def handle_call({:stream, topics, item}, _from, state) when is_list(topics) do
-    Enum.each(topics, fn t ->
-      do_stream(%{topic: t, item: item})
-    end)
-
-    {:reply, state, state}
-  end
-
-  def handle_call({:stream, topic, items}, _from, state) when is_list(items) do
-    Enum.each(items, fn i ->
-      do_stream(%{topic: topic, item: i})
-    end)
-
-    {:reply, state, state}
-  end
-
-  def handle_call({:stream, topic, item}, _from, state) do
-    do_stream(%{topic: topic, item: item})
-
-    {:reply, state, state}
-  end
-
-  defp do_stream(%{topic: "direct", item: item}) do
-    recipient_topics =
-      User.get_recipients_from_activity(item)
-      |> Enum.map(fn %{id: id} -> "direct:#{id}" end)
-
-    Enum.each(recipient_topics, fn user_topic ->
-      Logger.debug("Trying to push direct message to #{user_topic}\n\n")
-      push_to_socket(State.get_sockets(), user_topic, item)
-    end)
-  end
-
-  defp do_stream(%{topic: "participation", item: participation}) do
-    user_topic = "direct:#{participation.user_id}"
-    Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")
-
-    push_to_socket(State.get_sockets(), user_topic, participation)
-  end
-
-  defp do_stream(%{topic: "list", item: item}) do
-    # filter the recipient list if the activity is not public, see #270.
-    recipient_lists =
-      case Visibility.is_public?(item) do
-        true ->
-          Pleroma.List.get_lists_from_activity(item)
-
-        _ ->
-          Pleroma.List.get_lists_from_activity(item)
-          |> Enum.filter(fn list ->
-            owner = User.get_cached_by_id(list.user_id)
-
-            Visibility.visible_for_user?(item, owner)
-          end)
-      end
-
-    recipient_topics =
-      recipient_lists
-      |> Enum.map(fn %{id: id} -> "list:#{id}" end)
-
-    Enum.each(recipient_topics, fn list_topic ->
-      Logger.debug("Trying to push message to #{list_topic}\n\n")
-      push_to_socket(State.get_sockets(), list_topic, item)
-    end)
-  end
-
-  defp do_stream(%{topic: topic, item: %Notification{} = item})
-       when topic in ["user", "user:notification"] do
-    State.get_sockets()
-    |> Map.get("#{topic}:#{item.user_id}", [])
-    |> Enum.each(fn %StreamerSocket{transport_pid: transport_pid, user: socket_user} ->
-      with %User{} = user <- User.get_cached_by_ap_id(socket_user.ap_id),
-           true <- should_send?(user, item) do
-        send(transport_pid, {:text, StreamerView.render("notification.json", socket_user, item)})
-      end
-    end)
-  end
-
-  defp do_stream(%{topic: "user", item: item}) do
-    Logger.debug("Trying to push to users")
-
-    recipient_topics =
-      User.get_recipients_from_activity(item)
-      |> Enum.map(fn %{id: id} -> "user:#{id}" end)
-
-    Enum.each(recipient_topics, fn topic ->
-      push_to_socket(State.get_sockets(), topic, item)
-    end)
-  end
-
-  defp do_stream(%{topic: topic, item: item}) do
-    Logger.debug("Trying to push to #{topic}")
-    Logger.debug("Pushing item to #{topic}")
-    push_to_socket(State.get_sockets(), topic, item)
-  end
-
-  defp should_send?(%User{} = user, %Activity{} = item) do
-    %{block: blocked_ap_ids, mute: muted_ap_ids, reblog_mute: reblog_muted_ap_ids} =
-      User.outgoing_relationships_ap_ids(user, [:block, :mute, :reblog_mute])
-
-    recipient_blocks = MapSet.new(blocked_ap_ids ++ muted_ap_ids)
-    recipients = MapSet.new(item.recipients)
-    domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks)
-
-    with parent <- Object.normalize(item) || item,
-         true <-
-           Enum.all?([blocked_ap_ids, muted_ap_ids], &(item.actor not in &1)),
-         true <- item.data["type"] != "Announce" || item.actor not in reblog_muted_ap_ids,
-         true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(parent.data["actor"] not in &1)),
-         true <- MapSet.disjoint?(recipients, recipient_blocks),
-         %{host: item_host} <- URI.parse(item.actor),
-         %{host: parent_host} <- URI.parse(parent.data["actor"]),
-         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
-         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
-         true <- thread_containment(item, user),
-         false <- CommonAPI.thread_muted?(user, item) do
-      true
-    else
-      _ -> false
-    end
-  end
-
-  defp should_send?(%User{} = user, %Notification{activity: activity}) do
-    should_send?(user, activity)
-  end
-
-  def push_to_socket(topics, topic, %Participation{} = participation) do
-    Enum.each(topics[topic] || [], fn %StreamerSocket{transport_pid: transport_pid} ->
-      send(transport_pid, {:text, StreamerView.render("conversation.json", participation)})
-    end)
-  end
-
-  def push_to_socket(topics, topic, %Activity{
-        data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
-      }) do
-    Enum.each(topics[topic] || [], fn %StreamerSocket{transport_pid: transport_pid} ->
-      send(
-        transport_pid,
-        {:text, %{event: "delete", payload: to_string(deleted_activity_id)} |> Jason.encode!()}
-      )
-    end)
-  end
-
-  def push_to_socket(_topics, _topic, %Activity{data: %{"type" => "Delete"}}), do: :noop
-
-  def push_to_socket(topics, topic, item) do
-    Enum.each(topics[topic] || [], fn %StreamerSocket{
-                                        transport_pid: transport_pid,
-                                        user: socket_user
-                                      } ->
-      # Get the current user so we have up-to-date blocks etc.
-      if socket_user do
-        user = User.get_cached_by_ap_id(socket_user.ap_id)
-
-        if should_send?(user, item) do
-          send(transport_pid, {:text, StreamerView.render("update.json", item, user)})
-        end
-      else
-        send(transport_pid, {:text, StreamerView.render("update.json", item)})
-      end
-    end)
-  end
-
-  @spec thread_containment(Activity.t(), User.t()) :: boolean()
-  defp thread_containment(_activity, %User{skip_thread_containment: true}), do: true
-
-  defp thread_containment(activity, user) do
-    if Config.get([:instance, :skip_thread_containment]) do
-      true
-    else
-      ActivityPub.contain_activity(activity, user)
-    end
-  end
-end

lib/pleroma/web/templates/o_auth/mfa/recovery.html.eex

@@ -0,0 +1,24 @@
+<%= if get_flash(@conn, :info) do %>
+<p class="alert alert-info" role="alert"><%= get_flash(@conn, :info) %></p>
+<% end %>
+<%= if get_flash(@conn, :error) do %>
+<p class="alert alert-danger" role="alert"><%= get_flash(@conn, :error) %></p>
+<% end %>
+
+<h2>Two-factor recovery</h2>
+
+<%= form_for @conn, mfa_verify_path(@conn, :verify), [as: "mfa"], fn f -> %>
+<div class="input">
+  <%= label f, :code, "Recovery code" %>
+  <%= text_input f, :code %>
+  <%= hidden_input f, :mfa_token, value: @mfa_token %>
+  <%= hidden_input f, :state, value: @state %>
+  <%= hidden_input f, :redirect_uri, value: @redirect_uri %>
+  <%= hidden_input f, :challenge_type, value: "recovery" %>
+</div>
+
+<%= submit "Verify" %>
+<% end %>
+<a href="<%= mfa_path(@conn, :show, %{challenge_type: "totp", mfa_token: @mfa_token, state: @state, redirect_uri: @redirect_uri}) %>">
+  Enter a two-factor code
+</a>

lib/pleroma/web/templates/o_auth/mfa/totp.html.eex

@@ -0,0 +1,24 @@
+<%= if get_flash(@conn, :info) do %>
+<p class="alert alert-info" role="alert"><%= get_flash(@conn, :info) %></p>
+<% end %>
+<%= if get_flash(@conn, :error) do %>
+<p class="alert alert-danger" role="alert"><%= get_flash(@conn, :error) %></p>
+<% end %>
+
+<h2>Two-factor authentication</h2>
+
+<%= form_for @conn, mfa_verify_path(@conn, :verify), [as: "mfa"], fn f -> %>
+<div class="input">
+  <%= label f, :code, "Authentication code" %>
+  <%= text_input f, :code %>
+  <%= hidden_input f, :mfa_token, value: @mfa_token %>
+  <%= hidden_input f, :state, value: @state %>
+  <%= hidden_input f, :redirect_uri, value: @redirect_uri %>
+  <%= hidden_input f, :challenge_type, value: "totp" %>
+</div>
+
+<%= submit "Verify" %>
+<% end %>
+<a href="<%= mfa_path(@conn, :show, %{challenge_type: "recovery", mfa_token: @mfa_token, state: @state, redirect_uri: @redirect_uri}) %>">
+  Enter a two-factor recovery code
+</a>

lib/pleroma/web/templates/twitter_api/remote_follow/follow_mfa.html.eex

@@ -0,0 +1,13 @@
+<%= if @error do %>
+<h2><%= @error %></h2>
+<% end %>
+<h2>Two-factor authentication</h2>
+<p><%= @followee.nickname %></p>
+<img height="128" width="128" src="<%= avatar_url(@followee) %>">
+<%= form_for @conn, remote_follow_path(@conn, :do_follow), [as: "mfa"], fn f -> %>
+<%= text_input f, :code, placeholder: "Authentication code", required: true %>
+<br>
+<%= hidden_input f, :id, value: @followee.id %>
+<%= hidden_input f, :token, value: @mfa_token %>
+<%= submit "Authorize" %>
+<% end %>

lib/pleroma/web/twitter_api/controllers/remote_follow_controller.ex

@@ -8,10 +8,12 @@ defmodule Pleroma.Web.TwitterAPI.RemoteFollowController do
   require Logger
 
   alias Pleroma.Activity
+  alias Pleroma.MFA
   alias Pleroma.Object.Fetcher
   alias Pleroma.Plugs.OAuthScopesPlug
   alias Pleroma.User
   alias Pleroma.Web.Auth.Authenticator
+  alias Pleroma.Web.Auth.TOTPAuthenticator
   alias Pleroma.Web.CommonAPI
 
   @status_types ["Article", "Event", "Note", "Video", "Page", "Question"]
@@ -68,6 +70,8 @@ defmodule Pleroma.Web.TwitterAPI.RemoteFollowController do
 
   # POST  /ostatus_subscribe
   #
+  # adds a remote account in followers if user already is signed in.
+  #
   def do_follow(%{assigns: %{user: %User{} = user}} = conn, %{"user" => %{"id" => id}}) do
     with {:fetch_user, %User{} = followee} <- {:fetch_user, User.get_cached_by_id(id)},
          {:ok, _, _, _} <- CommonAPI.follow(user, followee) do
@@ -78,9 +82,33 @@ defmodule Pleroma.Web.TwitterAPI.RemoteFollowController do
     end
   end
 
+  # POST  /ostatus_subscribe
+  #
+  # step 1.
+  # checks login\password and displays step 2 form of MFA if need.
+  #
   def do_follow(conn, %{"authorization" => %{"name" => _, "password" => _, "id" => id}}) do
-    with {:fetch_user, %User{} = followee} <- {:fetch_user, User.get_cached_by_id(id)},
+    with {_, %User{} = followee} <- {:fetch_user, User.get_cached_by_id(id)},
          {_, {:ok, user}, _} <- {:auth, Authenticator.get_user(conn), followee},
+         {_, _, _, false} <- {:mfa_required, followee, user, MFA.require?(user)},
+         {:ok, _, _, _} <- CommonAPI.follow(user, followee) do
+      redirect(conn, to: "/users/#{followee.id}")
+    else
+      error ->
+        handle_follow_error(conn, error)
+    end
+  end
+
+  # POST  /ostatus_subscribe
+  #
+  # step 2
+  # checks TOTP code. otherwise displays form with errors
+  #
+  def do_follow(conn, %{"mfa" => %{"code" => code, "token" => token, "id" => id}}) do
+    with {_, %User{} = followee} <- {:fetch_user, User.get_cached_by_id(id)},
+         {_, _, {:ok, %{user: user}}} <- {:mfa_token, followee, MFA.Token.validate(token)},
+         {_, _, _, {:ok, _}} <-
+           {:verify_mfa_code, followee, token, TOTPAuthenticator.verify(code, user)},
          {:ok, _, _, _} <- CommonAPI.follow(user, followee) do
       redirect(conn, to: "/users/#{followee.id}")
     else
@@ -94,6 +122,23 @@ defmodule Pleroma.Web.TwitterAPI.RemoteFollowController do
     render(conn, "followed.html", %{error: "Insufficient permissions: follow | write:follows."})
   end
 
+  defp handle_follow_error(conn, {:mfa_token, followee, _} = _) do
+    render(conn, "follow_login.html", %{error: "Wrong username or password", followee: followee})
+  end
+
+  defp handle_follow_error(conn, {:verify_mfa_code, followee, token, _} = _) do
+    render(conn, "follow_mfa.html", %{
+      error: "Wrong authentication code",
+      followee: followee,
+      mfa_token: token
+    })
+  end
+
+  defp handle_follow_error(conn, {:mfa_required, followee, user, _} = _) do
+    {:ok, %{token: token}} = MFA.Token.create_token(user)
+    render(conn, "follow_mfa.html", %{followee: followee, mfa_token: token, error: false})
+  end
+
   defp handle_follow_error(conn, {:auth, _, followee} = _) do
     render(conn, "follow_login.html", %{error: "Wrong username or password", followee: followee})
   end