defmodule Pleroma.Web.ActivityPub.ActivityPub do
alias Pleroma.Activity
- alias Pleroma.Instances
+ alias Pleroma.Conversation
alias Pleroma.Notification
alias Pleroma.Object
+ alias Pleroma.Object.Fetcher
+ alias Pleroma.Pagination
alias Pleroma.Repo
alias Pleroma.Upload
alias Pleroma.User
alias Pleroma.Web.ActivityPub.MRF
alias Pleroma.Web.ActivityPub.Transmogrifier
- alias Pleroma.Web.Federator
- alias Pleroma.Web.OStatus
alias Pleroma.Web.WebFinger
import Ecto.Query
require Logger
- @httpoison Application.get_env(:pleroma, :httpoison)
-
# 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
if is_public?(object), do: User.decrease_note_count(actor), else: {:ok, actor}
end
- def insert(map, local \\ true) when is_map(map) do
+ def increase_replies_count_if_reply(%{
+ "object" => %{"inReplyTo" => reply_ap_id} = object,
+ "type" => "Create"
+ }) do
+ if is_public?(object) do
+ Object.increase_replies_count(reply_ap_id)
+ end
+ end
+
+ def increase_replies_count_if_reply(_create_data), do: :noop
+
+ def decrease_replies_count_if_reply(%Object{
+ data: %{"inReplyTo" => reply_ap_id} = object
+ }) do
+ if is_public?(object) do
+ Object.decrease_replies_count(reply_ap_id)
+ end
+ end
+
+ def decrease_replies_count_if_reply(_object), 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),
+ map <- lazy_put_activity_defaults(map, fake),
:ok <- check_actor_is_active(map["actor"]),
{_, true} <- {:remote_limit_error, check_remote_limit(map)},
{:ok, map} <- MRF.filter(map),
- {:ok, object} <- insert_full_object(map) do
- {recipients, _, _} = get_recipients(map)
-
+ {recipients, _, _} = get_recipients(map),
+ {:fake, false, map, recipients} <- {:fake, fake, map, recipients},
+ {:ok, map, object} <- insert_full_object(map) do
{:ok, activity} =
Repo.insert(%Activity{
data: map,
end)
Notification.create_notifications(activity)
+
+ participations =
+ activity
+ |> Conversation.create_or_bump_for()
+ |> get_participations()
+
stream_out(activity)
+ stream_out_participations(participations)
{:ok, activity}
else
- %Activity{} = activity -> {:ok, activity}
- error -> {:error, error}
+ %Activity{} = activity ->
+ {:ok, activity}
+
+ {:fake, true, map, recipients} ->
+ activity = %Activity{
+ data: map,
+ local: local,
+ actor: map["actor"],
+ recipients: recipients,
+ id: "pleroma:fakeid"
+ }
+
+ Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
+ {:ok, activity}
+
+ error ->
+ {:error, error}
end
end
+ defp get_participations({:ok, %{participations: participations}}), do: participations
+ defp get_participations(_), do: []
+
+ def stream_out_participations(participations) do
+ participations =
+ participations
+ |> Repo.preload(:user)
+
+ Enum.each(participations, fn participation ->
+ Pleroma.Web.Streamer.stream("participation", participation)
+ end)
+ end
+
def stream_out(activity) do
public = "https://www.w3.org/ns/activitystreams#Public"
end
if activity.data["type"] in ["Create"] do
- activity.data["object"]
+ object = Object.normalize(activity)
+
+ 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["object"]["attachment"] != [] do
+ if object.data["attachment"] != [] do
Pleroma.Web.Streamer.stream("public:media", activity)
if activity.local do
end
end
else
+ # TODO: Write test, replace with visibility test
if !Enum.member?(activity.data["cc"] || [], public) &&
!Enum.member?(
activity.data["to"],
- User.get_by_ap_id(activity.data["actor"]).follower_address
+ User.get_cached_by_ap_id(activity.data["actor"]).follower_address
),
do: Pleroma.Web.Streamer.stream("direct", activity)
end
end
end
- def create(%{to: to, actor: actor, context: context, object: object} = params) do
+ def create(%{to: to, actor: actor, context: context, object: object} = params, fake \\ false) do
additional = params[:additional] || %{}
# only accept false as false value
local = !(params[:local] == false)
%{to: to, actor: actor, published: published, context: context, object: object},
additional
),
- {:ok, activity} <- insert(create_data, local),
+ {:ok, activity} <- insert(create_data, local, fake),
+ {:fake, false, activity} <- {:fake, fake, activity},
+ _ <- increase_replies_count_if_reply(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),
:ok <- maybe_federate(activity) do
{:ok, activity}
+ else
+ {:fake, true, activity} ->
+ {:ok, activity}
end
end
"deleted_activity_id" => activity && activity.id
},
{:ok, activity} <- insert(data, local),
+ _ <- decrease_replies_count_if_reply(object),
# Changing note count prior to enqueuing federation task in order to avoid
# race conditions on updating user.info
{:ok, _actor} <- decrease_note_count_if_public(user, object),
:ok <- maybe_federate(activity) do
Enum.each(User.all_superusers(), fn superuser ->
superuser
- |> Pleroma.AdminEmail.report(actor, account, statuses, content)
- |> Pleroma.Mailer.deliver_async()
+ |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
+ |> Pleroma.Emails.Mailer.deliver_async()
end)
{:ok, activity}
end
end
- def fetch_activities_for_context(context, opts \\ %{}) do
+ defp fetch_activities_for_context_query(context, opts) do
public = ["https://www.w3.org/ns/activitystreams#Public"]
recipients =
if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
- query = from(activity in Activity)
-
- query =
- query
- |> restrict_blocked(opts)
- |> restrict_recipients(recipients, opts["user"])
-
- query =
- from(
- activity in query,
- where:
- fragment(
- "?->>'type' = ? and ?->>'context' = ?",
- activity.data,
- "Create",
- activity.data,
- ^context
- ),
- order_by: [desc: :id]
+ from(activity in Activity)
+ |> restrict_blocked(opts)
+ |> restrict_recipients(recipients, opts["user"])
+ |> where(
+ [activity],
+ fragment(
+ "?->>'type' = ? and ?->>'context' = ?",
+ activity.data,
+ "Create",
+ activity.data,
+ ^context
)
- |> Activity.with_preloaded_object()
+ )
+ |> order_by([activity], desc: activity.id)
+ end
- Repo.all(query)
+ @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
+ def fetch_activities_for_context(context, opts \\ %{}) do
+ context
+ |> fetch_activities_for_context_query(opts)
+ |> Activity.with_preloaded_object()
+ |> Repo.all()
+ end
+
+ @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
+ Pleroma.FlakeId.t() | nil
+ def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
+ context
+ |> fetch_activities_for_context_query(opts)
+ |> limit(1)
+ |> select([a], a.id)
+ |> Repo.one()
end
def fetch_public_activities(opts \\ %{}) do
q
|> restrict_unlisted()
- |> Repo.all()
+ |> Pagination.fetch_paginated(opts)
|> Enum.reverse()
end
defp restrict_since(query, _), do: query
+ defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
+ raise "Can't use the child object without preloading!"
+ end
+
defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
when is_list(tag_reject) and tag_reject != [] do
from(
- activity in query,
- where: fragment(~s(\(not \(? #> '{"object","tag"}'\) \\?| ?\)), activity.data, ^tag_reject)
+ [_activity, object] in query,
+ where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
)
end
defp restrict_tag_reject(query, _), do: query
+ defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
+ raise "Can't use the child object without preloading!"
+ end
+
defp restrict_tag_all(query, %{"tag_all" => tag_all})
when is_list(tag_all) and tag_all != [] do
from(
- activity in query,
- where: fragment(~s(\(? #> '{"object","tag"}'\) \\?& ?), activity.data, ^tag_all)
+ [_activity, object] in query,
+ where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
)
end
defp restrict_tag_all(query, _), do: query
+ defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
+ raise "Can't use the child object without preloading!"
+ end
+
defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
from(
- activity in query,
- where: fragment(~s(\(? #> '{"object","tag"}'\) \\?| ?), activity.data, ^tag)
+ [_activity, object] in query,
+ where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
)
end
defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
from(
- activity in query,
- where: fragment(~s(? <@ (? #> '{"object","tag"}'\)), ^tag, activity.data)
+ [_activity, object] in query,
+ where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
)
end
)
end
- defp restrict_limit(query, %{"limit" => limit}) do
- from(activity in query, limit: ^limit)
- end
-
- defp restrict_limit(query, _), do: query
-
defp restrict_local(query, %{"local_only" => true}) do
from(activity in query, where: activity.local == true)
end
defp restrict_local(query, _), do: query
- defp restrict_max(query, %{"max_id" => ""}), do: query
-
- defp restrict_max(query, %{"max_id" => max_id}) do
- from(activity in query, where: activity.id < ^max_id)
- end
-
- defp restrict_max(query, _), do: query
-
defp restrict_actor(query, %{"actor_id" => actor_id}) do
from(activity in query, where: activity.actor == ^actor_id)
end
defp restrict_favorited_by(query, _), do: query
+ defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
+ raise "Can't use the child object without preloading!"
+ end
+
defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
from(
- activity in query,
- where: fragment(~s(not (? #> '{"object","attachment"}' = ?\)), activity.data, ^[])
+ [_activity, object] in query,
+ where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
)
end
from(
activity in query,
where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
- where: fragment("not (?->'to' \\?| ?)", activity.data, ^blocks),
+ where: fragment("not (? && ?)", activity.recipients, ^blocks),
+ where:
+ fragment(
+ "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
+ activity.data,
+ activity.data,
+ ^blocks
+ ),
where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks)
)
end
from(
activity in query,
- where: fragment("not ?->>'type' = 'Announce'", activity.data),
- where: fragment("not ? = ANY(?)", activity.actor, ^muted_reblogs)
+ where:
+ fragment(
+ "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
+ activity.data,
+ activity.actor,
+ ^muted_reblogs
+ )
)
end
|> Activity.with_preloaded_object()
end
+ defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
+
+ defp maybe_preload_bookmarks(query, opts) do
+ query
+ |> Activity.with_preloaded_bookmark(opts["user"])
+ end
+
+ defp maybe_order(query, %{order: :desc}) do
+ query
+ |> order_by(desc: :id)
+ end
+
+ defp maybe_order(query, %{order: :asc}) do
+ query
+ |> order_by(asc: :id)
+ end
+
+ defp maybe_order(query, _), do: query
+
def fetch_activities_query(recipients, opts \\ %{}) do
- base_query =
- from(
- activity in Activity,
- limit: 20,
- order_by: [fragment("? desc nulls last", activity.id)]
- )
+ base_query = from(activity in Activity)
base_query
|> maybe_preload_objects(opts)
+ |> maybe_preload_bookmarks(opts)
+ |> maybe_order(opts)
|> restrict_recipients(recipients, opts["user"])
|> restrict_tag(opts)
|> restrict_tag_reject(opts)
|> restrict_tag_all(opts)
|> restrict_since(opts)
|> restrict_local(opts)
- |> restrict_limit(opts)
- |> restrict_max(opts)
|> restrict_actor(opts)
|> restrict_type(opts)
|> restrict_favorited_by(opts)
|> restrict_reblogs(opts)
|> restrict_pinned(opts)
|> restrict_muted_reblogs(opts)
+ |> Activity.restrict_deactivated_users()
end
def fetch_activities(recipients, opts \\ %{}) do
fetch_activities_query(recipients, opts)
- |> Repo.all()
+ |> Pagination.fetch_paginated(opts)
|> Enum.reverse()
end
def fetch_activities_bounded(recipients_to, recipients_cc, opts \\ %{}) do
fetch_activities_query([], opts)
|> restrict_to_cc(recipients_to, recipients_cc)
- |> Repo.all()
+ |> Pagination.fetch_paginated(opts)
|> Enum.reverse()
end
end
def fetch_and_prepare_user_from_ap_id(ap_id) do
- with {:ok, data} <- fetch_and_contain_remote_object_from_id(ap_id) do
+ with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id) do
user_data_from_user_object(data)
else
e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
end
def make_user_from_ap_id(ap_id) do
- if _user = User.get_by_ap_id(ap_id) do
+ if _user = User.get_cached_by_ap_id(ap_id) do
Transmogrifier.upgrade_user_from_ap_id(ap_id)
else
with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
end
end
- def should_federate?(inbox, public) do
- if public do
- true
- else
- inbox_info = URI.parse(inbox)
- !Enum.member?(Pleroma.Config.get([:instance, :quarantined_instances], []), inbox_info.host)
- end
- end
-
- def publish(actor, activity) do
- remote_followers =
- if actor.follower_address in activity.recipients do
- {:ok, followers} = User.get_followers(actor)
- followers |> Enum.filter(&(!&1.local))
- else
- []
- end
-
- public = is_public?(activity)
-
- {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
- json = Jason.encode!(data)
-
- (Pleroma.Web.Salmon.remote_users(activity) ++ remote_followers)
- |> Enum.filter(fn user -> User.ap_enabled?(user) end)
- |> Enum.map(fn %{info: %{source_data: data}} ->
- (is_map(data["endpoints"]) && Map.get(data["endpoints"], "sharedInbox")) || data["inbox"]
- end)
- |> Enum.uniq()
- |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
- |> Instances.filter_reachable()
- |> Enum.each(fn {inbox, unreachable_since} ->
- Federator.publish_single_ap(%{
- inbox: inbox,
- json: json,
- actor: actor,
- id: activity.data["id"],
- unreachable_since: unreachable_since
- })
- end)
- end
-
- def publish_one(%{inbox: inbox, json: json, actor: actor, id: id} = params) do
- Logger.info("Federating #{id} to #{inbox}")
- host = URI.parse(inbox).host
-
- digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64())
-
- date =
- NaiveDateTime.utc_now()
- |> Timex.format!("{WDshort}, {0D} {Mshort} {YYYY} {h24}:{m}:{s} GMT")
-
- signature =
- Pleroma.Web.HTTPSignatures.sign(actor, %{
- host: host,
- "content-length": byte_size(json),
- digest: digest,
- date: date
- })
-
- with {:ok, %{status: code}} when code in 200..299 <-
- result =
- @httpoison.post(
- inbox,
- json,
- [
- {"Content-Type", "application/activity+json"},
- {"Date", date},
- {"signature", signature},
- {"digest", digest}
- ]
- ) do
- if !Map.has_key?(params, :unreachable_since) || params[:unreachable_since],
- do: Instances.set_reachable(inbox)
-
- result
- else
- {_post_result, response} ->
- unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
- {:error, response}
- end
- end
-
- # TODO:
- # This will create a Create activity, which we need internally at the moment.
- def fetch_object_from_id(id) do
- if object = Object.get_cached_by_ap_id(id) do
- {:ok, object}
- else
- with {:ok, data} <- fetch_and_contain_remote_object_from_id(id),
- nil <- Object.normalize(data),
- params <- %{
- "type" => "Create",
- "to" => data["to"],
- "cc" => data["cc"],
- "actor" => data["actor"] || data["attributedTo"],
- "object" => data
- },
- :ok <- Transmogrifier.contain_origin(id, params),
- {:ok, activity} <- Transmogrifier.handle_incoming(params) do
- {:ok, Object.normalize(activity.data["object"])}
- else
- {:error, {:reject, nil}} ->
- {:reject, nil}
-
- object = %Object{} ->
- {:ok, object}
-
- _e ->
- Logger.info("Couldn't get object via AP, trying out OStatus fetching...")
-
- case OStatus.fetch_activity_from_url(id) do
- {:ok, [activity | _]} -> {:ok, Object.normalize(activity.data["object"])}
- e -> e
- end
- end
- end
- end
-
- def fetch_and_contain_remote_object_from_id(id) do
- Logger.info("Fetching object #{id} via AP")
-
- with true <- String.starts_with?(id, "http"),
- {:ok, %{body: body, status: code}} when code in 200..299 <-
- @httpoison.get(
- id,
- [{:Accept, "application/activity+json"}]
- ),
- {:ok, data} <- Jason.decode(body),
- :ok <- Transmogrifier.contain_origin_from_id(id, data) do
- {:ok, data}
- else
- e ->
- {:error, e}
- end
- end
-
# filter out broken threads
def contain_broken_threads(%Activity{} = activity, %User{} = user) do
entire_thread_visible_for_user?(activity, user)
projects / akkoma / blobdiff