X-Git-Url: http://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fweb%2Factivity_pub%2Factivity_pub.ex;h=8add62406a7a6ac0ab8fd053532b5ea1f1ad2661;hb=ad5263c647aea65dbeb4c329825671895e0a8863;hp=6c737d0a47302ac9903fd18d5c60dc92649e993d;hpb=81d1aa424d65b364ec8f2aee45247e7d95e3f255;p=akkoma diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex index 6c737d0a4..8add62406 100644 --- a/lib/pleroma/web/activity_pub/activity_pub.ex +++ b/lib/pleroma/web/activity_pub/activity_pub.ex @@ -5,7 +5,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do alias Pleroma.Activity alias Pleroma.Conversation - alias Pleroma.Instances alias Pleroma.Notification alias Pleroma.Object alias Pleroma.Object.Fetcher @@ -15,7 +14,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do alias Pleroma.User alias Pleroma.Web.ActivityPub.MRF alias Pleroma.Web.ActivityPub.Transmogrifier - alias Pleroma.Web.Federator alias Pleroma.Web.WebFinger import Ecto.Query @@ -24,8 +22,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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 @@ -137,9 +133,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do activity end - Task.start(fn -> - Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity) - end) + PleromaJobQueue.enqueue(:background, Pleroma.Web.RichMedia.Helpers, [:fetch, activity]) Notification.create_notifications(activity) @@ -405,16 +399,12 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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 = Pleroma.Config.get([:activitypub, :outgoing_blocks]) + unfollow_blocked = Pleroma.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, @@ -546,8 +536,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do ) ) - Ecto.Adapters.SQL.to_sql(:all, Repo, query) - query else Logger.error("Could not restrict visibility to #{visibility}") @@ -563,8 +551,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility) ) - Ecto.Adapters.SQL.to_sql(:all, Repo, query) - query end @@ -575,6 +561,18 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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) + ) + + query + end + + defp restrict_thread_visibility(query, _), do: query + def fetch_user_activities(user, reading_user, params \\ %{}) do params = params @@ -651,20 +649,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub 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 @@ -701,6 +685,12 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub 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, @@ -756,8 +746,11 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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: @@ -767,7 +760,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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 @@ -815,11 +809,40 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do |> 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_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) + 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) base_query |> 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_tag_reject(opts) @@ -828,15 +851,18 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do |> 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_replies(opts) |> restrict_reblogs(opts) |> restrict_pinned(opts) |> restrict_muted_reblogs(opts) + |> Activity.restrict_deactivated_users() end def fetch_activities(recipients, opts \\ %{}) do @@ -845,9 +871,18 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do |> Enum.reverse() end - def fetch_activities_bounded(recipients_to, recipients_cc, opts \\ %{}) do + 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) + ) + end + + 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 @@ -865,7 +900,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do end end - def user_data_from_user_object(data) do + defp object_to_user_data(data) do avatar = data["icon"]["url"] && %{ @@ -912,9 +947,19 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do {:ok, user_data} 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) do + {:ok, data} else e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}") end @@ -940,89 +985,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub 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 - # filter out broken threads def contain_broken_threads(%Activity{} = activity, %User{} = user) do entire_thread_visible_for_user?(activity, user) @@ -1033,11 +995,10 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do contain_broken_threads(activity, user) end - # do post-processing on a timeline - def contain_timeline(timeline, user) do - timeline - |> Enum.filter(fn activity -> - contain_activity(activity, user) - end) + def fetch_direct_messages_query do + Activity + |> restrict_type(%{"type" => "Create"}) + |> restrict_visibility(%{visibility: "direct"}) + |> order_by([activity], asc: activity.id) end end