# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.ActivityPub.ActivityPub do
- alias Pleroma.{Activity, Repo, Object, Upload, User, Notification}
- alias Pleroma.Web.ActivityPub.{Transmogrifier, MRF}
+ alias Pleroma.Activity
+ alias Pleroma.Repo
+ alias Pleroma.Object
+ alias Pleroma.Upload
+ alias Pleroma.User
+ alias Pleroma.Notification
+ alias Pleroma.Instances
+ alias Pleroma.Web.ActivityPub.Transmogrifier
+ alias Pleroma.Web.ActivityPub.MRF
alias Pleroma.Web.WebFinger
alias Pleroma.Web.Federator
alias Pleroma.Web.OStatus
+
import Ecto.Query
import Pleroma.Web.ActivityPub.Utils
+ import Pleroma.Web.ActivityPub.Visibility
+
require Logger
@httpoison Application.get_env(:pleroma, :httpoison)
defp get_recipients(%{"type" => "Announce"} = data) do
to = data["to"] || []
cc = data["cc"] || []
- recipients = to ++ cc
actor = User.get_cached_by_ap_id(data["actor"])
- recipients
- |> Enum.filter(fn recipient ->
- case User.get_cached_by_ap_id(recipient) do
- nil ->
- true
+ recipients =
+ (to ++ cc)
+ |> Enum.filter(fn recipient ->
+ case User.get_cached_by_ap_id(recipient) do
+ nil ->
+ true
+
+ user ->
+ User.following?(user, actor)
+ end
+ end)
- user ->
- User.following?(user, actor)
- end
- end)
+ {recipients, to, cc}
+ end
+ defp get_recipients(%{"type" => "Create"} = data) do
+ to = data["to"] || []
+ cc = data["cc"] || []
+ actor = data["actor"] || []
+ recipients = (to ++ cc ++ [actor]) |> Enum.uniq()
{recipients, to, cc}
end
end
end
- defp check_remote_limit(%{"object" => %{"content" => content}}) do
+ defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
limit = Pleroma.Config.get([:instance, :remote_limit])
String.length(content) <= limit
end
defp check_remote_limit(_), do: true
+ def increase_note_count_if_public(actor, object) do
+ if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
+ end
+
+ def decrease_note_count_if_public(actor, object) 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
with nil <- Activity.normalize(map),
map <- lazy_put_activity_defaults(map),
recipients: recipients
})
+ Task.start(fn ->
+ Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
+ end)
+
Notification.create_notifications(activity)
stream_out(activity)
{:ok, activity}
activity.data["object"]
|> Map.get("tag", [])
|> Enum.filter(fn tag -> is_bitstring(tag) end)
- |> Enum.map(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end)
+ |> Enum.each(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end)
if activity.data["object"]["attachment"] != [] do
Pleroma.Web.Streamer.stream("public:media", activity)
additional
),
{:ok, activity} <- insert(create_data, local),
- :ok <- maybe_federate(activity),
- {:ok, _actor} <- User.increase_note_count(actor) do
+ # 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}
end
end
# only accept false as false value
local = !(params[:local] == false)
- with data <- %{"to" => to, "type" => "Accept", "actor" => actor, "object" => object},
+ with data <- %{"to" => to, "type" => "Accept", "actor" => actor.ap_id, "object" => object},
{:ok, activity} <- insert(data, local),
:ok <- maybe_federate(activity) do
{:ok, activity}
# only accept false as false value
local = !(params[:local] == false)
- with data <- %{"to" => to, "type" => "Reject", "actor" => actor, "object" => object},
+ with data <- %{"to" => to, "type" => "Reject", "actor" => actor.ap_id, "object" => object},
{:ok, activity} <- insert(data, local),
:ok <- maybe_federate(activity) do
{:ok, activity}
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"] || [])
data = %{
"type" => "Delete",
"actor" => actor,
"object" => id,
- "to" => [user.follower_address, "https://www.w3.org/ns/activitystreams#Public"]
+ "to" => to
}
with {:ok, _} <- Object.delete(object),
{:ok, activity} <- insert(data, local),
- :ok <- maybe_federate(activity),
- {:ok, _actor} <- User.decrease_note_count(user) do
+ # 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
{:ok, activity}
end
end
end
end
+ def flag(
+ %{
+ actor: actor,
+ context: context,
+ account: account,
+ statuses: statuses,
+ content: content
+ } = params
+ ) do
+ additional = params[:additional] || %{}
+
+ # only accept false as false value
+ local = !(params[:local] == false)
+
+ %{
+ actor: actor,
+ context: context,
+ account: account,
+ statuses: statuses,
+ content: content
+ }
+ |> make_flag_data(additional)
+ |> insert(local)
+ end
+
def fetch_activities_for_context(context, opts \\ %{}) do
public = ["https://www.w3.org/ns/activitystreams#Public"]
@valid_visibilities ~w[direct unlisted public private]
+ defp restrict_visibility(query, %{visibility: visibility})
+ when is_list(visibility) do
+ if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
+ query =
+ from(
+ a in query,
+ where:
+ fragment(
+ "activity_visibility(?, ?, ?) = ANY (?)",
+ a.actor,
+ a.recipients,
+ a.data,
+ ^visibility
+ )
+ )
+
+ Ecto.Adapters.SQL.to_sql(:all, Repo, query)
+
+ query
+ else
+ Logger.error("Could not restrict visibility to #{visibility}")
+ end
+ end
+
defp restrict_visibility(query, %{visibility: visibility})
when visibility in @valid_visibilities do
query =
|> Enum.reverse()
end
+ defp restrict_since(query, %{"since_id" => ""}), do: query
+
defp restrict_since(query, %{"since_id" => since_id}) do
from(activity in query, where: activity.id > ^since_id)
end
defp restrict_since(query, _), do: query
- defp restrict_tag(query, %{"tag" => tag}) do
+ defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
+ when is_list(tag_reject) and tag_reject != [] do
+ from(
+ activity in query,
+ where: fragment("(not (? #> '{\"object\",\"tag\"}') \\?| ?)", activity.data, ^tag_reject)
+ )
+ end
+
+ defp restrict_tag_reject(query, _), do: query
+
+ defp restrict_tag_all(query, %{"tag_all" => tag_all})
+ when is_list(tag_all) and tag_all != [] do
+ from(
+ activity in query,
+ where: fragment("(? #> '{\"object\",\"tag\"}') \\?& ?", activity.data, ^tag_all)
+ )
+ end
+
+ defp restrict_tag_all(query, _), do: query
+
+ defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
+ from(
+ activity in query,
+ where: fragment("(? #> '{\"object\",\"tag\"}') \\?| ?", activity.data, ^tag)
+ )
+ end
+
+ defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
from(
activity in query,
where: fragment("? <@ (? #> '{\"object\",\"tag\"}')", ^tag, activity.data)
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_actor(query, _), do: query
defp restrict_type(query, %{"type" => type}) when is_binary(type) do
- restrict_type(query, %{"type" => [type]})
+ from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
end
defp restrict_type(query, %{"type" => type}) do
defp restrict_reblogs(query, _), do: query
+ 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
+ mutes = info.mutes
+
+ from(
+ activity in query,
+ where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
+ where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
+ )
+ end
+
+ defp restrict_muted(query, _), do: query
+
defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
blocks = info.blocks || []
domain_blocks = info.domain_blocks || []
base_query
|> 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_type(opts)
|> restrict_favorited_by(opts)
|> restrict_blocked(opts)
+ |> restrict_muted(opts)
|> restrict_media(opts)
|> restrict_visibility(opts)
|> restrict_replies(opts)
end
def publish(actor, activity) do
- followers =
+ remote_followers =
if actor.follower_address in activity.recipients do
{:ok, followers} = User.get_followers(actor)
followers |> Enum.filter(&(!&1.local))
public = is_public?(activity)
- remote_inboxes =
- (Pleroma.Web.Salmon.remote_users(activity) ++ 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)
-
{:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
json = Jason.encode!(data)
- Enum.each(remote_inboxes, fn inbox ->
- Federator.enqueue(:publish_single_ap, %{
+ (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"]
+ id: activity.data["id"],
+ unreachable_since: unreachable_since
})
end)
end
- def publish_one(%{inbox: inbox, json: json, actor: actor, id: id}) do
+ 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
+ digest: digest,
+ date: date
})
- @httpoison.post(
- inbox,
- json,
- [
- {"Content-Type", "application/activity+json"},
- {"signature", signature},
- {"digest", digest}
- ]
- )
+ 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:
if object = Object.get_cached_by_ap_id(id) do
{:ok, object}
else
- Logger.info("Fetching #{id} via AP")
-
with {:ok, data} <- fetch_and_contain_remote_object_from_id(id),
nil <- Object.normalize(data),
params <- %{
end
def fetch_and_contain_remote_object_from_id(id) do
- Logger.info("Fetching #{id} via AP")
+ Logger.info("Fetching object #{id} via AP")
with true <- String.starts_with?(id, "http"),
{:ok, %{body: body, status: code}} when code in 200..299 <-
end
end
- def is_public?(%Object{data: %{"type" => "Tombstone"}}), do: false
- def is_public?(%Object{data: data}), do: is_public?(data)
- def is_public?(%Activity{data: data}), do: is_public?(data)
-
- def is_public?(data) do
- "https://www.w3.org/ns/activitystreams#Public" in (data["to"] ++ (data["cc"] || []))
- end
-
- def visible_for_user?(activity, nil) do
- is_public?(activity)
- end
-
- def visible_for_user?(activity, user) do
- x = [user.ap_id | user.following]
- y = activity.data["to"] ++ (activity.data["cc"] || [])
- visible_for_user?(activity, nil) || Enum.any?(x, &(&1 in y))
- end
-
- # guard
- def entire_thread_visible_for_user?(nil, _user), do: false
-
- # child
- def entire_thread_visible_for_user?(
- %Activity{data: %{"object" => %{"inReplyTo" => parent_id}}} = tail,
- user
- )
- when is_binary(parent_id) do
- parent = Activity.get_in_reply_to_activity(tail)
- visible_for_user?(tail, user) && entire_thread_visible_for_user?(parent, user)
- end
-
- # root
- def entire_thread_visible_for_user?(tail, user), do: visible_for_user?(tail, user)
-
# filter out broken threads
def contain_broken_threads(%Activity{} = activity, %User{} = user) do
entire_thread_visible_for_user?(activity, user)
projects / akkoma / blobdiff