defmodule Pleroma.Web.ActivityPub.ActivityPub do
alias Pleroma.Activity
+ alias Pleroma.Config
alias Pleroma.Conversation
alias Pleroma.Notification
alias Pleroma.Object
+ alias Pleroma.Object.Containment
alias Pleroma.Object.Fetcher
alias Pleroma.Pagination
alias Pleroma.Repo
import Pleroma.Web.ActivityPub.Visibility
require Logger
+ require Pleroma.Constants
# For Announce activities, we filter the recipients based on following status for any actors
# that match actual users. See issue #164 for more information about why this is necessary.
defp get_recipients(%{"type" => "Announce"} = data) do
- to = data["to"] || []
- cc = data["cc"] || []
+ to = Map.get(data, "to", [])
+ cc = Map.get(data, "cc", [])
+ bcc = Map.get(data, "bcc", [])
actor = User.get_cached_by_ap_id(data["actor"])
recipients =
- (to ++ cc)
- |> Enum.filter(fn recipient ->
+ Enum.filter(Enum.concat([to, cc, bcc]), fn recipient ->
case User.get_cached_by_ap_id(recipient) do
- nil ->
- true
-
- user ->
- User.following?(user, actor)
+ nil -> true
+ user -> User.following?(user, actor)
end
end)
end
defp get_recipients(%{"type" => "Create"} = data) do
- to = data["to"] || []
- cc = data["cc"] || []
- actor = data["actor"] || []
- recipients = (to ++ cc ++ [actor]) |> Enum.uniq()
+ to = Map.get(data, "to", [])
+ cc = Map.get(data, "cc", [])
+ bcc = Map.get(data, "bcc", [])
+ actor = Map.get(data, "actor", [])
+ recipients = [to, cc, bcc, [actor]] |> Enum.concat() |> Enum.uniq()
{recipients, to, cc}
end
defp get_recipients(data) do
- to = data["to"] || []
- cc = data["cc"] || []
- recipients = to ++ cc
+ to = Map.get(data, "to", [])
+ cc = Map.get(data, "cc", [])
+ bcc = Map.get(data, "bcc", [])
+ recipients = Enum.concat([to, cc, bcc])
{recipients, to, cc}
end
end
defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
- limit = Pleroma.Config.get([:instance, :remote_limit])
+ limit = Config.get([:instance, :remote_limit])
String.length(content) <= limit
end
def decrease_replies_count_if_reply(_object), do: :noop
+ def increase_poll_votes_if_vote(%{
+ "object" => %{"inReplyTo" => reply_ap_id, "name" => name},
+ "type" => "Create"
+ }) do
+ Object.increase_vote_count(reply_ap_id, name)
+ end
+
+ def increase_poll_votes_if_vote(_create_data), do: :noop
+
def insert(map, local \\ true, fake \\ false) when is_map(map) do
with nil <- Activity.normalize(map),
map <- lazy_put_activity_defaults(map, fake),
{:ok, map} <- MRF.filter(map),
{recipients, _, _} = get_recipients(map),
{:fake, false, map, recipients} <- {:fake, fake, map, recipients},
+ :ok <- Containment.contain_child(map),
{:ok, map, object} <- insert_full_object(map) do
{:ok, activity} =
Repo.insert(%Activity{
end)
end
- def stream_out(activity) do
- public = "https://www.w3.org/ns/activitystreams#Public"
+ def stream_out_participations(%Object{data: %{"context" => context}}, user) do
+ with %Conversation{} = conversation <- Conversation.get_for_ap_id(context),
+ conversation = Repo.preload(conversation, :participations),
+ last_activity_id =
+ fetch_latest_activity_id_for_context(conversation.ap_id, %{
+ "user" => user,
+ "blocking_user" => user
+ }) do
+ if last_activity_id do
+ stream_out_participations(conversation.participations)
+ end
+ end
+ end
- if activity.data["type"] in ["Create", "Announce", "Delete"] do
- Pleroma.Web.Streamer.stream("user", activity)
- Pleroma.Web.Streamer.stream("list", activity)
+ def stream_out_participations(_, _), do: :noop
- if Enum.member?(activity.data["to"], public) do
- Pleroma.Web.Streamer.stream("public", activity)
+ def stream_out(activity) do
+ if activity.data["type"] in ["Create", "Announce", "Delete"] do
+ object = Object.normalize(activity)
+ # Do not stream out poll replies
+ unless object.data["type"] == "Answer" do
+ Pleroma.Web.Streamer.stream("user", activity)
+ Pleroma.Web.Streamer.stream("list", activity)
- if activity.local do
- Pleroma.Web.Streamer.stream("public:local", activity)
- end
+ if get_visibility(activity) == "public" do
+ Pleroma.Web.Streamer.stream("public", activity)
- if activity.data["type"] in ["Create"] do
- object = Object.normalize(activity)
+ if activity.local do
+ Pleroma.Web.Streamer.stream("public:local", activity)
+ end
- object.data
- |> Map.get("tag", [])
- |> Enum.filter(fn tag -> is_bitstring(tag) end)
- |> Enum.each(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end)
+ if activity.data["type"] in ["Create"] do
+ object.data
+ |> Map.get("tag", [])
+ |> Enum.filter(fn tag -> is_bitstring(tag) end)
+ |> Enum.each(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end)
- if object.data["attachment"] != [] do
- Pleroma.Web.Streamer.stream("public:media", activity)
+ if object.data["attachment"] != [] do
+ Pleroma.Web.Streamer.stream("public:media", activity)
- if activity.local do
- Pleroma.Web.Streamer.stream("public:local:media", activity)
+ if activity.local do
+ Pleroma.Web.Streamer.stream("public:local:media", activity)
+ end
end
end
+ else
+ if get_visibility(activity) == "direct",
+ do: Pleroma.Web.Streamer.stream("direct", activity)
end
- else
- # TODO: Write test, replace with visibility test
- if !Enum.member?(activity.data["cc"] || [], public) &&
- !Enum.member?(
- activity.data["to"],
- User.get_cached_by_ap_id(activity.data["actor"]).follower_address
- ),
- do: Pleroma.Web.Streamer.stream("direct", activity)
end
end
end
{: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),
# Changing note count prior to enqueuing federation task in order to avoid
# race conditions on updating user.info
{:ok, _actor} <- increase_note_count_if_public(actor, activity),
else
{:fake, true, activity} ->
{:ok, activity}
+
+ {:error, message} ->
+ {:error, message}
end
end
end
end
+ def delete(%User{ap_id: ap_id, follower_address: follower_address} = user) do
+ with data <- %{
+ "to" => [follower_address],
+ "type" => "Delete",
+ "actor" => ap_id,
+ "object" => %{"type" => "Person", "id" => ap_id}
+ },
+ {:ok, activity} <- insert(data, true, true),
+ :ok <- maybe_federate(activity) do
+ {:ok, user}
+ end
+ end
+
def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, local \\ true) do
user = User.get_cached_by_ap_id(actor)
to = (object.data["to"] || []) ++ (object.data["cc"] || [])
"to" => to,
"deleted_activity_id" => activity && activity.id
},
- {:ok, activity} <- insert(data, local),
+ {:ok, activity} <- insert(data, local, false),
+ stream_out_participations(object, user),
_ <- decrease_replies_count_if_reply(object),
# Changing note count prior to enqueuing federation task in order to avoid
# race conditions on updating user.info
end
def block(blocker, blocked, activity_id \\ nil, local \\ true) do
- ap_config = Application.get_env(:pleroma, :activitypub)
- unfollow_blocked = Keyword.get(ap_config, :unfollow_blocked)
- outgoing_blocks = Keyword.get(ap_config, :outgoing_blocks)
+ outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
+ unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
- with true <- unfollow_blocked do
+ if unfollow_blocked do
follow_activity = fetch_latest_follow(blocker, blocked)
-
- if follow_activity do
- unfollow(blocker, blocked, nil, local)
- end
+ if follow_activity, do: unfollow(blocker, blocked, nil, local)
end
with true <- outgoing_blocks,
end
defp fetch_activities_for_context_query(context, opts) do
- public = ["https://www.w3.org/ns/activitystreams#Public"]
+ public = [Pleroma.Constants.as_public()]
recipients =
if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
from(activity in Activity)
+ |> maybe_preload_objects(opts)
|> restrict_blocked(opts)
|> restrict_recipients(recipients, opts["user"])
|> where(
^context
)
)
+ |> exclude_poll_votes(opts)
|> order_by([activity], desc: activity.id)
end
def fetch_activities_for_context(context, opts \\ %{}) do
context
|> fetch_activities_for_context_query(opts)
- |> Activity.with_preloaded_object()
|> Repo.all()
end
Pleroma.FlakeId.t() | nil
def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
context
- |> fetch_activities_for_context_query(opts)
+ |> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
|> limit(1)
|> select([a], a.id)
|> Repo.one()
end
def fetch_public_activities(opts \\ %{}) do
- q = fetch_activities_query(["https://www.w3.org/ns/activitystreams#Public"], opts)
+ q = fetch_activities_query([Pleroma.Constants.as_public()], opts)
q
|> restrict_unlisted()
defp restrict_visibility(query, %{visibility: visibility})
when visibility in @valid_visibilities do
- query =
- from(
- a in query,
- where:
- fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
- )
-
- query
+ from(
+ a in query,
+ where:
+ fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
+ )
end
defp restrict_visibility(_query, %{visibility: visibility})
defp restrict_visibility(query, _visibility), do: query
- defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}) do
- query =
- from(
- a in query,
- where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
- )
+ defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
+ do: query
- query
+ defp restrict_thread_visibility(
+ query,
+ %{"user" => %User{info: %{skip_thread_containment: true}}},
+ _
+ ),
+ do: query
+
+ defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
+ from(
+ a in query,
+ where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
+ )
end
- defp restrict_thread_visibility(query, _), do: query
+ defp restrict_thread_visibility(query, _, _), do: query
def fetch_user_activities(user, reading_user, params \\ %{}) do
params =
|> Map.put("pinned_activity_ids", user.info.pinned_activities)
recipients =
- if reading_user do
- ["https://www.w3.org/ns/activitystreams#Public"] ++
- [reading_user.ap_id | reading_user.following]
- else
- ["https://www.w3.org/ns/activitystreams#Public"]
- end
+ user_activities_recipients(%{
+ "godmode" => params["godmode"],
+ "reading_user" => reading_user
+ })
fetch_activities(recipients, params)
|> Enum.reverse()
end
+ defp user_activities_recipients(%{"godmode" => true}) do
+ []
+ end
+
+ defp user_activities_recipients(%{"reading_user" => reading_user}) do
+ if reading_user do
+ [Pleroma.Constants.as_public()] ++ [reading_user.ap_id | reading_user.following]
+ else
+ [Pleroma.Constants.as_public()]
+ end
+ end
+
defp restrict_since(query, %{"since_id" => ""}), do: query
defp restrict_since(query, %{"since_id" => since_id}) do
defp restrict_tag(query, _), do: query
- defp restrict_to_cc(query, recipients_to, recipients_cc) do
- from(
- activity in query,
- where:
- fragment(
- "(?->'to' \\?| ?) or (?->'cc' \\?| ?)",
- activity.data,
- ^recipients_to,
- activity.data,
- ^recipients_cc
- )
- )
- end
-
defp restrict_recipients(query, [], _user), do: query
defp restrict_recipients(query, recipients, nil) do
defp restrict_type(query, _), do: query
+ defp restrict_state(query, %{"state" => state}) do
+ from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
+ end
+
+ defp restrict_state(query, _), do: query
+
defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
from(
activity in query,
blocks = info.blocks || []
domain_blocks = info.domain_blocks || []
+ query =
+ if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
+
from(
- activity in query,
+ [activity, object: o] in query,
where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
where: fragment("not (? && ?)", activity.recipients, ^blocks),
where:
activity.data,
^blocks
),
- where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks)
+ where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks),
+ where: fragment("not (split_part(?->>'actor', '/', 3) = ANY(?))", o.data, ^domain_blocks)
)
end
fragment(
"not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
activity.data,
- ^["https://www.w3.org/ns/activitystreams#Public"]
+ ^[Pleroma.Constants.as_public()]
)
)
end
defp restrict_muted_reblogs(query, _), do: query
+ defp exclude_poll_votes(query, %{"include_poll_votes" => "true"}), do: query
+
+ defp exclude_poll_votes(query, _) do
+ if has_named_binding?(query, :object) do
+ from([activity, object: o] in query,
+ where: fragment("not(?->>'type' = ?)", o.data, "Answer")
+ )
+ else
+ query
+ end
+ end
+
defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
defp maybe_preload_objects(query, _) do
|> Activity.with_preloaded_bookmark(opts["user"])
end
+ defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
+
+ defp maybe_set_thread_muted_field(query, opts) do
+ query
+ |> Activity.with_set_thread_muted_field(opts["user"])
+ end
+
defp maybe_order(query, %{order: :desc}) do
query
|> order_by(desc: :id)
defp maybe_order(query, _), do: query
def fetch_activities_query(recipients, opts \\ %{}) do
- base_query = from(activity in Activity)
+ config = %{
+ skip_thread_containment: Config.get([:instance, :skip_thread_containment])
+ }
- base_query
+ Activity
|> maybe_preload_objects(opts)
|> maybe_preload_bookmarks(opts)
+ |> maybe_set_thread_muted_field(opts)
|> maybe_order(opts)
|> restrict_recipients(recipients, opts["user"])
|> restrict_tag(opts)
|> restrict_local(opts)
|> restrict_actor(opts)
|> restrict_type(opts)
+ |> restrict_state(opts)
|> restrict_favorited_by(opts)
|> restrict_blocked(opts)
|> restrict_muted(opts)
|> restrict_media(opts)
|> restrict_visibility(opts)
- |> restrict_thread_visibility(opts)
+ |> restrict_thread_visibility(opts, config)
|> restrict_replies(opts)
|> restrict_reblogs(opts)
|> restrict_pinned(opts)
|> restrict_muted_reblogs(opts)
|> Activity.restrict_deactivated_users()
+ |> exclude_poll_votes(opts)
end
def fetch_activities(recipients, opts \\ %{}) do
- fetch_activities_query(recipients, opts)
+ list_memberships = Pleroma.List.memberships(opts["user"])
+
+ fetch_activities_query(recipients ++ list_memberships, opts)
|> Pagination.fetch_paginated(opts)
|> Enum.reverse()
+ |> maybe_update_cc(list_memberships, opts["user"])
+ end
+
+ defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
+ when is_list(list_memberships) and length(list_memberships) > 0 do
+ Enum.map(activities, fn
+ %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
+ if Enum.any?(bcc, &(&1 in list_memberships)) do
+ update_in(activity.data["cc"], &[user_ap_id | &1])
+ else
+ activity
+ end
+
+ activity ->
+ activity
+ end)
+ end
+
+ defp maybe_update_cc(activities, _, _), do: activities
+
+ def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
+ from(activity in query,
+ where:
+ fragment("? && ?", activity.recipients, ^recipients) or
+ (fragment("? && ?", activity.recipients, ^recipients_with_public) and
+ ^Pleroma.Constants.as_public() in activity.recipients)
+ )
end
- def fetch_activities_bounded(recipients_to, recipients_cc, opts \\ %{}) do
+ def fetch_activities_bounded(recipients, recipients_with_public, opts \\ %{}) do
fetch_activities_query([], opts)
- |> restrict_to_cc(recipients_to, recipients_cc)
+ |> fetch_activities_bounded_query(recipients, recipients_with_public)
|> Pagination.fetch_paginated(opts)
|> Enum.reverse()
end
end
end
- def user_data_from_user_object(data) do
+ defp object_to_user_data(data) do
avatar =
data["icon"]["url"] &&
%{
user_data = %{
ap_id: data["id"],
info: %{
- "ap_enabled" => true,
- "source_data" => data,
- "banner" => banner,
- "locked" => locked
+ ap_enabled: true,
+ source_data: data,
+ banner: banner,
+ locked: locked
},
avatar: avatar,
name: data["name"],
follower_address: data["followers"],
+ following_address: data["following"],
bio: data["summary"]
}
{:ok, user_data}
end
+ def fetch_follow_information_for_user(user) do
+ with {:ok, following_data} <-
+ Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
+ following_count when is_integer(following_count) <- following_data["totalItems"],
+ {:ok, hide_follows} <- collection_private(following_data),
+ {:ok, followers_data} <-
+ Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
+ followers_count when is_integer(followers_count) <- followers_data["totalItems"],
+ {:ok, hide_followers} <- collection_private(followers_data) do
+ {:ok,
+ %{
+ hide_follows: hide_follows,
+ follower_count: followers_count,
+ following_count: following_count,
+ hide_followers: hide_followers
+ }}
+ else
+ {:error, _} = e ->
+ e
+
+ e ->
+ {:error, e}
+ end
+ end
+
+ defp maybe_update_follow_information(data) do
+ with {:enabled, true} <-
+ {:enabled, Pleroma.Config.get([:instance, :external_user_synchronization])},
+ {:ok, info} <- fetch_follow_information_for_user(data) do
+ info = Map.merge(data.info, info)
+ Map.put(data, :info, info)
+ else
+ {:enabled, false} ->
+ data
+
+ e ->
+ Logger.error(
+ "Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
+ )
+
+ data
+ end
+ end
+
+ defp collection_private(data) do
+ if is_map(data["first"]) and
+ data["first"]["type"] in ["CollectionPage", "OrderedCollectionPage"] do
+ {:ok, false}
+ else
+ with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
+ Fetcher.fetch_and_contain_remote_object_from_id(data["first"]) do
+ {:ok, false}
+ else
+ {:error, {:ok, %{status: code}}} when code in [401, 403] ->
+ {:ok, true}
+
+ {:error, _} = e ->
+ e
+
+ e ->
+ {:error, e}
+ end
+ end
+ end
+
+ def user_data_from_user_object(data) do
+ with {:ok, data} <- MRF.filter(data),
+ {:ok, data} <- object_to_user_data(data) do
+ {:ok, data}
+ else
+ e -> {:error, e}
+ 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) do
- user_data_from_user_object(data)
+ with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
+ {:ok, data} <- user_data_from_user_object(data),
+ data <- maybe_update_follow_information(data) do
+ {:ok, data}
else
e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
end