defmodule Pleroma.Web.ActivityPub.ActivityPub do
alias Pleroma.Activity
+ alias Pleroma.Activity.Ir.Topics
+ 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
alias Pleroma.User
alias Pleroma.Web.ActivityPub.MRF
alias Pleroma.Web.ActivityPub.Transmogrifier
+ alias Pleroma.Web.Streamer
alias Pleroma.Web.WebFinger
+ alias Pleroma.Workers.BackgroundWorker
import Ecto.Query
import Pleroma.Web.ActivityPub.Utils
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
if not is_nil(actor) do
with user <- User.get_cached_by_ap_id(actor),
false <- user.info.deactivated do
- :ok
+ true
else
- _e -> :reject
+ _e -> false
end
else
- :ok
+ true
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 increase_poll_votes_if_vote(_create_data), do: :noop
- def insert(map, local \\ true, fake \\ false) when is_map(map) do
+ 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),
- :ok <- check_actor_is_active(map["actor"]),
+ true <- bypass_actor_check || check_actor_is_active(map["actor"]),
{_, true} <- {:remote_limit_error, check_remote_limit(map)},
{: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{
# Splice in the child object if we have one.
activity =
- if !is_nil(object) do
+ if not is_nil(object) do
Map.put(activity, :object, object)
else
activity
end
- PleromaJobQueue.enqueue(:background, Pleroma.Web.RichMedia.Helpers, [:fetch, activity])
+ BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})
Notification.create_notifications(activity)
participations
|> Repo.preload(:user)
- Enum.each(participations, fn participation ->
- Pleroma.Web.Streamer.stream("participation", participation)
- end)
+ Streamer.stream("participation", participations)
end
- def stream_out(activity) do
- public = "https://www.w3.org/ns/activitystreams#Public"
-
- 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 Enum.member?(activity.data["to"], public) do
- Pleroma.Web.Streamer.stream("public", activity)
-
- if activity.local do
- Pleroma.Web.Streamer.stream("public:local", activity)
- end
+ 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"] 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)
+ def stream_out_participations(_, _), do: :noop
- if object.data["attachment"] != [] do
- Pleroma.Web.Streamer.stream("public:media", activity)
+ def stream_out(%Activity{data: %{"type" => data_type}} = activity)
+ when data_type in ["Create", "Announce", "Delete"] do
+ activity
+ |> Topics.get_activity_topics()
+ |> Streamer.stream(activity)
+ end
- if activity.local do
- Pleroma.Web.Streamer.stream("public:local:media", activity)
- end
- end
- 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
+ def stream_out(_activity) do
+ :noop
end
def create(%{to: to, actor: actor, context: context, object: object} = params, fake \\ false) do
else
{:fake, true, activity} ->
{:ok, activity}
+
+ {:error, message} ->
+ {:error, message}
end
end
end
end
- def unlike(
- %User{} = actor,
- %Object{} = object,
- activity_id \\ nil,
- local \\ true
- ) do
+ def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
unlike_data <- make_unlike_data(actor, like_activity, activity_id),
{:ok, unlike_activity} <- insert(unlike_data, local),
def follow(follower, followed, activity_id \\ nil, local \\ true) do
with data <- make_follow_data(follower, followed, activity_id),
{:ok, activity} <- insert(data, local),
- :ok <- maybe_federate(activity) do
+ :ok <- maybe_federate(activity),
+ _ <- User.set_follow_state_cache(follower.ap_id, followed.ap_id, activity.data["state"]) do
{:ok, activity}
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, 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
end
+ @spec block(User.t(), User.t(), String.t() | nil, boolean) :: {:ok, Activity.t() | nil}
def block(blocker, blocked, activity_id \\ nil, local \\ true) do
- outgoing_blocks = Pleroma.Config.get([:activitypub, :outgoing_blocks])
- unfollow_blocked = Pleroma.Config.get([:activitypub, :unfollow_blocked])
+ outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
+ unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
if unfollow_blocked do
follow_activity = fetch_latest_follow(blocker, blocked)
end
end
+ @spec flag(map()) :: {:ok, Activity.t()} | any
def flag(
%{
actor: actor,
- context: context,
+ context: _context,
account: account,
statuses: statuses,
content: content
additional = params[:additional] || %{}
- params = %{
- actor: actor,
- context: context,
- account: account,
- statuses: statuses,
- content: content
- }
-
additional =
if forward do
Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
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)
- |> exclude_poll_votes(opts)
+ |> maybe_preload_bookmarks(opts)
+ |> maybe_set_thread_muted_field(opts)
|> restrict_blocked(opts)
|> restrict_recipients(recipients, opts["user"])
|> where(
^context
)
)
+ |> exclude_poll_votes(opts)
+ |> exclude_id(opts)
|> order_by([activity], desc: activity.id)
end
end
@spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
- Pleroma.FlakeId.t() | nil
+ FlakeId.Ecto.CompatType.t() | nil
def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
context
|> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
|> Repo.one()
end
- def fetch_public_activities(opts \\ %{}) do
- q = fetch_activities_query(["https://www.w3.org/ns/activitystreams#Public"], opts)
+ def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
+ opts = Map.drop(opts, ["user"])
- q
+ [Pleroma.Constants.as_public()]
+ |> fetch_activities_query(opts)
|> restrict_unlisted()
- |> Pagination.fetch_paginated(opts)
+ |> Pagination.fetch_paginated(opts, pagination)
|> Enum.reverse()
end
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 =
params
|> Map.put("type", ["Create", "Announce"])
+ |> Map.put("user", reading_user)
|> Map.put("actor_id", user.ap_id)
|> Map.put("whole_db", true)
|> 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_favorited_by(query, %{"favorited_by" => ap_id}) do
from(
- activity in query,
- where: fragment(~s(? <@ (? #> '{"object","likes"}'\)), ^ap_id, activity.data)
+ [_activity, object] in query,
+ where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
)
end
defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
- defp restrict_muted(query, %{"muting_user" => %User{info: info}}) do
+ defp restrict_muted(query, %{"muting_user" => %User{info: info}} = opts) do
mutes = info.mutes
- from(
- activity in query,
- where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
- where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
- )
+ query =
+ from([activity] in query,
+ where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
+ where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
+ )
+
+ unless opts["skip_preload"] do
+ from([thread_mute: tm] in query, where: is_nil(tm.user_id))
+ else
+ query
+ end
end
defp restrict_muted(query, _), do: query
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, %{"include_poll_votes" => true}), do: query
defp exclude_poll_votes(query, _) do
if has_named_binding?(query, :object) do
end
end
+ defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
+ from(activity in query, where: activity.id != ^id)
+ end
+
+ defp exclude_id(query, _), do: query
+
defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
defp maybe_preload_objects(query, _) do
defp maybe_set_thread_muted_field(query, opts) do
query
- |> Activity.with_set_thread_muted_field(opts["user"])
+ |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
end
defp maybe_order(query, %{order: :desc}) do
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)
|> 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)
|> exclude_poll_votes(opts)
end
- def fetch_activities(recipients, opts \\ %{}) do
- fetch_activities_query(recipients, opts)
- |> Pagination.fetch_paginated(opts)
+ def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
+ list_memberships = Pleroma.List.memberships(opts["user"])
+
+ fetch_activities_query(recipients ++ list_memberships, opts)
+ |> Pagination.fetch_paginated(opts, pagination)
|> 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
- "https://www.w3.org/ns/activitystreams#Public" in activity.recipients)
+ ^Pleroma.Constants.as_public() in activity.recipients)
)
end
- def fetch_activities_bounded(recipients, recipients_with_public, opts \\ %{}) do
+ def fetch_activities_bounded(
+ recipients,
+ recipients_with_public,
+ opts \\ %{},
+ pagination \\ :keyset
+ ) do
fetch_activities_query([], opts)
|> fetch_activities_bounded_query(recipients, recipients_with_public)
- |> Pagination.fetch_paginated(opts)
+ |> Pagination.fetch_paginated(opts, pagination)
|> Enum.reverse()
end
"url" => [%{"href" => data["image"]["url"]}]
}
+ fields =
+ data
+ |> Map.get("attachment", [])
+ |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
+ |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
+
locked = data["manuallyApprovesFollowers"] || false
data = Transmogrifier.maybe_fix_user_object(data)
+ discoverable = data["discoverable"] || false
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,
+ fields: fields,
+ locked: locked,
+ discoverable: discoverable
},
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
def fetch_and_prepare_user_from_ap_id(ap_id) do
with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
- {:ok, data} <- user_data_from_user_object(data) do
+ {: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)}")