X-Git-Url: http://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fweb%2Factivity_pub%2Factivity_pub.ex;h=501c8b845656b747cc99c4827e6b3d8cfbb5a610;hb=305d2194136d4560e02c110d528164034d3727b2;hp=c5f62c4f8c5702156e6efca2059bfb577d95607f;hpb=0c750bc432a6329be7f0dc15707e46625b11faf3;p=akkoma diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex index c5f62c4f8..501c8b845 100644 --- a/lib/pleroma/web/activity_pub/activity_pub.ex +++ b/lib/pleroma/web/activity_pub/activity_pub.ex @@ -3,13 +3,22 @@ # 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 + require Logger @httpoison Application.get_env(:pleroma, :httpoison) @@ -19,20 +28,28 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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 @@ -56,7 +73,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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 @@ -80,6 +97,10 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do recipients: recipients }) + Task.start(fn -> + Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity) + end) + Notification.create_notifications(activity) stream_out(activity) {:ok, activity} @@ -92,7 +113,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do def stream_out(activity) do public = "https://www.w3.org/ns/activitystreams#Public" - if activity.data["type"] in ["Create", "Announce"] do + if activity.data["type"] in ["Create", "Announce", "Delete"] do Pleroma.Web.Streamer.stream("user", activity) Pleroma.Web.Streamer.stream("list", activity) @@ -103,16 +124,18 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do Pleroma.Web.Streamer.stream("public:local", activity) end - 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) + if activity.data["type"] in ["Create"] do + activity.data["object"] + |> 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 - Pleroma.Web.Streamer.stream("public:media", activity) + if activity.data["object"]["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 @@ -138,8 +161,9 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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} <- User.increase_note_count(actor), + :ok <- maybe_federate(activity) do {:ok, activity} end end @@ -224,10 +248,11 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do %User{ap_id: _} = user, %Object{data: %{"id" => _}} = object, activity_id \\ nil, - local \\ true + local \\ true, + public \\ true ) do with true <- is_public?(object), - announce_data <- make_announce_data(user, object, activity_id), + announce_data <- make_announce_data(user, object, activity_id, public), {:ok, activity} <- insert(announce_data, local), {:ok, object} <- add_announce_to_object(activity, object), :ok <- maybe_federate(activity) do @@ -285,8 +310,9 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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} <- User.decrease_note_count(user), + :ok <- maybe_federate(activity) do {:ok, activity} end end @@ -364,21 +390,18 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do @valid_visibilities ~w[direct unlisted public private] - defp restrict_visibility(query, %{visibility: "direct"}) do - public = "https://www.w3.org/ns/activitystreams#Public" + 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) + ) - from( - activity in query, - join: sender in User, - on: sender.ap_id == activity.actor, - # Are non-direct statuses with no to/cc possible? - where: - fragment( - "not (? && ?)", - [^public, sender.follower_address], - activity.recipients - ) - ) + Ecto.Adapters.SQL.to_sql(:all, Repo, query) + + query end defp restrict_visibility(_query, %{visibility: visibility}) @@ -408,13 +431,42 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do |> 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) @@ -463,6 +515,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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 @@ -476,7 +530,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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 @@ -518,15 +572,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do defp restrict_reblogs(query, _), do: query - # Only search through last 100_000 activities by default - defp restrict_recent(query, %{"whole_db" => true}), do: query - - defp restrict_recent(query, _) do - since = (Repo.aggregate(Activity, :max, :id) || 0) - 100_000 - - from(activity in query, where: activity.id > ^since) - end - defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do blocks = info.blocks || [] domain_blocks = info.domain_blocks || [] @@ -570,6 +615,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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) @@ -577,7 +624,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do |> restrict_actor(opts) |> restrict_type(opts) |> restrict_favorited_by(opts) - |> restrict_recent(opts) |> restrict_blocked(opts) |> restrict_media(opts) |> restrict_visibility(opts) @@ -697,7 +743,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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)) @@ -707,29 +753,29 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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 @@ -742,15 +788,26 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do digest: digest }) - @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"}, + {"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: @@ -809,13 +866,24 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do end end - def is_public?(%Object{data: %{"type" => "Tombstone"}}) do - false + 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?(%{"directMessage" => true}), do: false + + def is_public?(data) do + "https://www.w3.org/ns/activitystreams#Public" in (data["to"] ++ (data["cc"] || [])) end - def is_public?(activity) do - "https://www.w3.org/ns/activitystreams#Public" in (activity.data["to"] ++ - (activity.data["cc"] || [])) + def is_private?(activity) do + !is_public?(activity) && Enum.any?(activity.data["to"], &String.contains?(&1, "/followers")) + end + + def is_direct?(%Activity{data: %{"directMessage" => true}}), do: true + def is_direct?(%Object{data: %{"directMessage" => true}}), do: true + + def is_direct?(activity) do + !is_public?(activity) && !is_private?(activity) end def visible_for_user?(activity, nil) do