defmodule Pleroma.Web.ActivityPub.ActivityPub do
alias Pleroma.Activity
alias Pleroma.Activity.Ir.Topics
- alias Pleroma.ActivityExpiration
alias Pleroma.Config
alias Pleroma.Constants
alias Pleroma.Conversation
alias Pleroma.Conversation.Participation
+ alias Pleroma.Filter
alias Pleroma.Maps
alias Pleroma.Notification
alias Pleroma.Object
defp check_remote_limit(_), do: true
- defp increase_note_count_if_public(actor, object) do
+ def increase_note_count_if_public(actor, object) do
if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
end
defp increase_replies_count_if_reply(_create_data), do: :noop
- defp increase_poll_votes_if_vote(%{
- "object" => %{"inReplyTo" => reply_ap_id, "name" => name},
- "type" => "Create",
- "actor" => actor
- }) do
- Object.increase_vote_count(reply_ap_id, name, actor)
- end
-
- defp increase_poll_votes_if_vote(_create_data), do: :noop
-
- @object_types ["ChatMessage"]
+ @object_types ~w[ChatMessage Question Answer Audio Video Event Article]
@spec persist(map(), keyword()) :: {:ok, Activity.t() | Object.t()}
def persist(%{"type" => type} = object, meta) when type in @object_types do
with {:ok, object} <- Object.create(object) do
local: local,
recipients: recipients,
actor: object["actor"]
- }) do
+ }),
+ # TODO: add tests for expired activities, when Note type will be supported in new pipeline
+ {:ok, _} <- maybe_create_activity_expiration(activity) do
{:ok, activity, meta}
end
end
def insert(map, local \\ true, fake \\ false, bypass_actor_check \\ false) when is_map(map) do
with nil <- Activity.normalize(map),
map <- lazy_put_activity_defaults(map, fake),
- true <- bypass_actor_check || check_actor_is_active(map["actor"]),
- {_, true} <- {:remote_limit_error, check_remote_limit(map)},
+ {_, true} <- {:actor_check, bypass_actor_check || check_actor_is_active(map["actor"])},
+ {_, true} <- {:remote_limit_pass, check_remote_limit(map)},
{:ok, map} <- MRF.filter(map),
{recipients, _, _} = get_recipients(map),
{:fake, false, map, recipients} <- {:fake, fake, map, recipients},
{:containment, :ok} <- {:containment, Containment.contain_child(map)},
- {:ok, map, object} <- insert_full_object(map) do
- {:ok, activity} =
- %Activity{
- data: map,
- local: local,
- actor: map["actor"],
- recipients: recipients
- }
- |> Repo.insert()
- |> maybe_create_activity_expiration()
-
+ {:ok, map, object} <- insert_full_object(map),
+ {:ok, activity} <- insert_activity_with_expiration(map, local, recipients) do
# Splice in the child object if we have one.
activity = Maps.put_if_present(activity, :object, object)
%Activity{} = activity ->
{:ok, activity}
+ {:actor_check, _} ->
+ {:error, false}
+
+ {:containment, _} = error ->
+ error
+
+ {:error, _} = error ->
+ error
+
{:fake, true, map, recipients} ->
activity = %Activity{
data: map,
Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
{:ok, activity}
- error ->
- {:error, error}
+ {:remote_limit_pass, _} ->
+ {:error, :remote_limit}
+
+ {:reject, _} = e ->
+ {:error, e}
+ end
+ end
+
+ defp insert_activity_with_expiration(data, local, recipients) do
+ struct = %Activity{
+ data: data,
+ local: local,
+ actor: data["actor"],
+ recipients: recipients
+ }
+
+ with {:ok, activity} <- Repo.insert(struct) do
+ maybe_create_activity_expiration(activity)
end
end
stream_out_participations(participations)
end
- defp maybe_create_activity_expiration({:ok, %{data: %{"expires_at" => expires_at}} = activity}) do
- with {:ok, _} <- ActivityExpiration.create(activity, expires_at) do
+ defp maybe_create_activity_expiration(
+ %{data: %{"expires_at" => %DateTime{} = expires_at}} = activity
+ ) do
+ with {:ok, _job} <-
+ Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
+ activity_id: activity.id,
+ expires_at: expires_at
+ }) do
{:ok, activity}
end
end
- defp maybe_create_activity_expiration(result), do: result
+ defp maybe_create_activity_expiration(activity), do: {:ok, activity}
defp create_or_bump_conversation(activity, actor) do
with {:ok, conversation} <- Conversation.create_or_bump_for(activity),
with {:ok, activity} <- insert(create_data, local, fake),
{:fake, false, activity} <- {:fake, fake, activity},
_ <- increase_replies_count_if_reply(create_data),
- _ <- 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),
end
end
- @spec accept(map()) :: {:ok, Activity.t()} | {:error, any()}
- def accept(params) do
- accept_or_reject("Accept", params)
- end
-
- @spec reject(map()) :: {:ok, Activity.t()} | {:error, any()}
- def reject(params) do
- accept_or_reject("Reject", params)
- end
-
- @spec accept_or_reject(String.t(), map()) :: {:ok, Activity.t()} | {:error, any()}
- defp accept_or_reject(type, %{to: to, actor: actor, object: object} = params) do
- local = Map.get(params, :local, true)
- activity_id = Map.get(params, :activity_id, nil)
-
- data =
- %{"to" => to, "type" => type, "actor" => actor.ap_id, "object" => object}
- |> Maps.put_if_present("id", activity_id)
-
- with {:ok, activity} <- insert(data, local),
- _ <- notify_and_stream(activity),
- :ok <- maybe_federate(activity) do
- {:ok, activity}
- end
- end
-
- @spec follow(User.t(), User.t(), String.t() | nil, boolean(), keyword()) ::
- {:ok, Activity.t()} | {:error, any()}
- def follow(follower, followed, activity_id \\ nil, local \\ true, opts \\ []) do
- with {:ok, result} <-
- Repo.transaction(fn -> do_follow(follower, followed, activity_id, local, opts) end) do
- result
- end
- end
-
- defp do_follow(follower, followed, activity_id, local, opts) do
- skip_notify_and_stream = Keyword.get(opts, :skip_notify_and_stream, false)
- data = make_follow_data(follower, followed, activity_id)
-
- with {:ok, activity} <- insert(data, local),
- _ <- skip_notify_and_stream || notify_and_stream(activity),
- :ok <- maybe_federate(activity) do
- {:ok, activity}
- else
- {:error, error} -> Repo.rollback(error)
- end
- end
-
@spec unfollow(User.t(), User.t(), String.t() | nil, boolean()) ::
{:ok, Activity.t()} | nil | {:error, any()}
def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
end
end
- @spec block(User.t(), User.t(), String.t() | nil, boolean()) ::
- {:ok, Activity.t()} | {:error, any()}
- def block(blocker, blocked, activity_id \\ nil, local \\ true) do
- with {:ok, result} <-
- Repo.transaction(fn -> do_block(blocker, blocked, activity_id, local) end) do
- result
- end
- end
-
- defp do_block(blocker, blocked, activity_id, local) do
- unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
-
- if unfollow_blocked and fetch_latest_follow(blocker, blocked) do
- unfollow(blocker, blocked, nil, local)
- end
-
- block_data = make_block_data(blocker, blocked, activity_id)
-
- with {:ok, activity} <- insert(block_data, local),
- _ <- notify_and_stream(activity),
- :ok <- maybe_federate(activity) do
- {:ok, activity}
- else
- {:error, error} -> Repo.rollback(error)
- end
- end
-
@spec flag(map()) :: {:ok, Activity.t()} | {:error, any()}
def flag(
%{
|> maybe_set_thread_muted_field(opts)
|> restrict_blocked(opts)
|> restrict_recipients(recipients, opts[:user])
+ |> restrict_filtered(opts)
|> where(
[activity],
fragment(
end
defp restrict_replies(query, %{
- reply_filtering_user: user,
+ reply_filtering_user: %User{} = user,
reply_visibility: "self"
}) do
from(
end
defp restrict_replies(query, %{
- reply_filtering_user: user,
+ reply_filtering_user: %User{} = user,
reply_visibility: "following"
}) do
from(
[activity, object] in query,
where:
fragment(
- "?->>'inReplyTo' is null OR ? && array_remove(?, ?) OR ? = ?",
+ """
+ ?->>'type' != 'Create' -- This isn't a Create
+ OR ?->>'inReplyTo' is null -- this isn't a reply
+ OR ? && array_remove(?, ?) -- The recipient is us or one of our friends,
+ -- unless they are the author (because authors
+ -- are also part of the recipients). This leads
+ -- to a bug that self-replies by friends won't
+ -- show up.
+ OR ? = ? -- The actor is us
+ """,
+ activity.data,
object.data,
^[user.ap_id | User.get_cached_user_friends_ap_ids(user)],
activity.recipients,
query =
from([activity] in query,
where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
- where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
+ where:
+ fragment(
+ "not (?->'to' \\?| ?) or ? = ?",
+ activity.data,
+ ^mutes,
+ activity.actor,
+ ^user.ap_id
+ )
)
unless opts[:skip_preload] do
from(
[activity, object: o] in query,
where: fragment("not (? = ANY(?))", activity.actor, ^blocked_ap_ids),
- where: fragment("not (? && ?)", activity.recipients, ^blocked_ap_ids),
+ where:
+ fragment(
+ "((not (? && ?)) or ? = ?)",
+ activity.recipients,
+ ^blocked_ap_ids,
+ activity.actor,
+ ^user.ap_id
+ ),
where:
fragment(
"recipients_contain_blocked_domains(?, ?) = false",
defp restrict_muted_reblogs(query, _), do: query
- defp restrict_instance(query, %{instance: instance}) do
- users =
- from(
- u in User,
- select: u.ap_id,
- where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
- )
- |> Repo.all()
-
- from(activity in query, where: activity.actor in ^users)
+ defp restrict_instance(query, %{instance: instance}) when is_binary(instance) do
+ from(
+ activity in query,
+ where: fragment("split_part(actor::text, '/'::text, 3) = ?", ^instance)
+ )
end
defp restrict_instance(query, _), do: query
+ defp restrict_filtered(query, %{user: %User{} = user}) do
+ case Filter.compose_regex(user) do
+ nil ->
+ query
+
+ regex ->
+ from([activity, object] in query,
+ where:
+ fragment("not(?->>'content' ~* ?)", object.data, ^regex) or
+ activity.actor == ^user.ap_id
+ )
+ end
+ end
+
+ defp restrict_filtered(query, %{blocking_user: %User{} = user}) do
+ restrict_filtered(query, %{user: user})
+ end
+
+ defp restrict_filtered(query, _), do: query
+
defp exclude_poll_votes(query, %{include_poll_votes: true}), do: query
defp exclude_poll_votes(query, _) do
|> restrict_favorited_by(opts)
|> restrict_blocked(restrict_blocked_opts)
|> restrict_muted(restrict_muted_opts)
+ |> restrict_filtered(opts)
|> restrict_media(opts)
|> restrict_visibility(opts)
|> restrict_thread_visibility(opts, config)
|> restrict_muted_reblogs(restrict_muted_reblogs_opts)
|> restrict_instance(opts)
|> restrict_announce_object_actor(opts)
+ |> restrict_filtered(opts)
|> Activity.restrict_deactivated_users()
|> exclude_poll_votes(opts)
|> exclude_chat_messages(opts)
{String.trim(name, ":"), url}
end)
- locked = data["manuallyApprovesFollowers"] || false
+ is_locked = data["manuallyApprovesFollowers"] || false
+ capabilities = data["capabilities"] || %{}
+ accepts_chat_messages = capabilities["acceptsChatMessages"]
data = Transmogrifier.maybe_fix_user_object(data)
- discoverable = data["discoverable"] || false
+ is_discoverable = data["discoverable"] || false
invisible = data["invisible"] || false
actor_type = data["type"] || "Person"
banner: banner,
fields: fields,
emoji: emojis,
- locked: locked,
- discoverable: discoverable,
+ is_locked: is_locked,
+ is_discoverable: is_discoverable,
invisible: invisible,
avatar: avatar,
name: data["name"],
follower_address: data["followers"],
following_address: data["following"],
- bio: data["summary"],
+ bio: data["summary"] || "",
actor_type: actor_type,
also_known_as: Map.get(data, "alsoKnownAs", []),
public_key: public_key,
inbox: data["inbox"],
- shared_inbox: shared_inbox
+ shared_inbox: shared_inbox,
+ accepts_chat_messages: accepts_chat_messages
}
# nickname can be nil because of virtual actors
def fetch_follow_information_for_user(user) do
with {:ok, following_data} <-
- Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
+ Fetcher.fetch_and_contain_remote_object_from_id(user.following_address,
+ force_http: true
+ ),
{:ok, hide_follows} <- collection_private(following_data),
{:ok, followers_data} <-
- Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
+ Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address, force_http: true),
{:ok, hide_followers} <- collection_private(followers_data) do
{:ok,
%{
end
end
- def fetch_and_prepare_user_from_ap_id(ap_id) do
- with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
+ def fetch_and_prepare_user_from_ap_id(ap_id, opts \\ []) do
+ with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id, opts),
{:ok, data} <- user_data_from_user_object(data) do
{:ok, maybe_update_follow_information(data)}
else
+ # If this has been deleted, only log a debug and not an error
{:error, "Object has been deleted" = e} ->
Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
{:error, e}
+ {:error, {:reject, reason} = e} ->
+ Logger.info("Rejected user #{ap_id}: #{inspect(reason)}")
+ {:error, e}
+
{:error, e} ->
Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
{:error, e}
end
end
- def make_user_from_ap_id(ap_id) do
+ def maybe_handle_clashing_nickname(data) do
+ with nickname when is_binary(nickname) <- data[:nickname],
+ %User{} = old_user <- User.get_by_nickname(nickname),
+ {_, false} <- {:ap_id_comparison, data[:ap_id] == old_user.ap_id} do
+ Logger.info(
+ "Found an old user for #{nickname}, the old ap id is #{old_user.ap_id}, new one is #{
+ data[:ap_id]
+ }, renaming."
+ )
+
+ old_user
+ |> User.remote_user_changeset(%{nickname: "#{old_user.id}.#{old_user.nickname}"})
+ |> User.update_and_set_cache()
+ else
+ {:ap_id_comparison, true} ->
+ Logger.info(
+ "Found an old user for #{data[:nickname]}, but the ap id #{data[:ap_id]} is the same as the new user. Race condition? Not changing anything."
+ )
+
+ _ ->
+ nil
+ end
+ end
+
+ def make_user_from_ap_id(ap_id, opts \\ []) do
user = User.get_cached_by_ap_id(ap_id)
if user && !User.ap_enabled?(user) do
Transmogrifier.upgrade_user_from_ap_id(ap_id)
else
- with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
+ with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id, opts) do
if user do
user
|> User.remote_user_changeset(data)
|> User.update_and_set_cache()
else
+ maybe_handle_clashing_nickname(data)
+
data
|> User.remote_user_changeset()
|> Repo.insert()