git.squeep.com Git - akkoma/blob - lib/pleroma/web/activity_pub/activity_pub.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Web.ActivityPub.ActivityPub do
   6   alias Pleroma.Activity
   7   alias Pleroma.Conversation
   8   alias Pleroma.Instances
   9   alias Pleroma.Notification
  10   alias Pleroma.Object
  11   alias Pleroma.Object.Fetcher
  12   alias Pleroma.Pagination
  13   alias Pleroma.Repo
  14   alias Pleroma.Upload
  15   alias Pleroma.User
  16   alias Pleroma.Web.ActivityPub.MRF
  17   alias Pleroma.Web.ActivityPub.Transmogrifier
  18   alias Pleroma.Web.Federator
  19   alias Pleroma.Web.WebFinger
  20
  21   import Ecto.Query
  22   import Pleroma.Web.ActivityPub.Utils
  23   import Pleroma.Web.ActivityPub.Visibility
  24
  25   require Logger
  26
  27   @httpoison Application.get_env(:pleroma, :httpoison)
  28
  29   # For Announce activities, we filter the recipients based on following status for any actors
  30   # that match actual users.  See issue #164 for more information about why this is necessary.
  31   defp get_recipients(%{"type" => "Announce"} = data) do
  32     to = data["to"] || []
  33     cc = data["cc"] || []
  34     actor = User.get_cached_by_ap_id(data["actor"])
  35
  36     recipients =
  37       (to ++ cc)
  38       |> Enum.filter(fn recipient ->
  39         case User.get_cached_by_ap_id(recipient) do
  40           nil ->
  41             true
  42
  43           user ->
  44             User.following?(user, actor)
  45         end
  46       end)
  47
  48     {recipients, to, cc}
  49   end
  50
  51   defp get_recipients(%{"type" => "Create"} = data) do
  52     to = data["to"] || []
  53     cc = data["cc"] || []
  54     actor = data["actor"] || []
  55     recipients = (to ++ cc ++ [actor]) |> Enum.uniq()
  56     {recipients, to, cc}
  57   end
  58
  59   defp get_recipients(data) do
  60     to = data["to"] || []
  61     cc = data["cc"] || []
  62     recipients = to ++ cc
  63     {recipients, to, cc}
  64   end
  65
  66   defp check_actor_is_active(actor) do
  67     if not is_nil(actor) do
  68       with user <- User.get_cached_by_ap_id(actor),
  69            false <- user.info.deactivated do
  70         :ok
  71       else
  72         _e -> :reject
  73       end
  74     else
  75       :ok
  76     end
  77   end
  78
  79   defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
  80     limit = Pleroma.Config.get([:instance, :remote_limit])
  81     String.length(content) <= limit
  82   end
  83
  84   defp check_remote_limit(_), do: true
  85
  86   def increase_note_count_if_public(actor, object) do
  87     if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
  88   end
  89
  90   def decrease_note_count_if_public(actor, object) do
  91     if is_public?(object), do: User.decrease_note_count(actor), else: {:ok, actor}
  92   end
  93
  94   def increase_replies_count_if_reply(%{
  95         "object" => %{"inReplyTo" => reply_ap_id} = object,
  96         "type" => "Create"
  97       }) do
  98     if is_public?(object) do
  99       Object.increase_replies_count(reply_ap_id)
 100     end
 101   end
 102
 103   def increase_replies_count_if_reply(_create_data), do: :noop
 104
 105   def decrease_replies_count_if_reply(%Object{
 106         data: %{"inReplyTo" => reply_ap_id} = object
 107       }) do
 108     if is_public?(object) do
 109       Object.decrease_replies_count(reply_ap_id)
 110     end
 111   end
 112
 113   def decrease_replies_count_if_reply(_object), do: :noop
 114
 115   def insert(map, local \\ true, fake \\ false) when is_map(map) do
 116     with nil <- Activity.normalize(map),
 117          map <- lazy_put_activity_defaults(map, fake),
 118          :ok <- check_actor_is_active(map["actor"]),
 119          {_, true} <- {:remote_limit_error, check_remote_limit(map)},
 120          {:ok, map} <- MRF.filter(map),
 121          {recipients, _, _} = get_recipients(map),
 122          {:fake, false, map, recipients} <- {:fake, fake, map, recipients},
 123          {:ok, map, object} <- insert_full_object(map) do
 124       {:ok, activity} =
 125         Repo.insert(%Activity{
 126           data: map,
 127           local: local,
 128           actor: map["actor"],
 129           recipients: recipients
 130         })
 131
 132       # Splice in the child object if we have one.
 133       activity =
 134         if !is_nil(object) do
 135           Map.put(activity, :object, object)
 136         else
 137           activity
 138         end
 139
 140       Task.start(fn ->
 141         Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
 142       end)
 143
 144       Notification.create_notifications(activity)
 145
 146       participations =
 147         activity
 148         |> Conversation.create_or_bump_for()
 149         |> get_participations()
 150
 151       stream_out(activity)
 152       stream_out_participations(participations)
 153       {:ok, activity}
 154     else
 155       %Activity{} = activity ->
 156         {:ok, activity}
 157
 158       {:fake, true, map, recipients} ->
 159         activity = %Activity{
 160           data: map,
 161           local: local,
 162           actor: map["actor"],
 163           recipients: recipients,
 164           id: "pleroma:fakeid"
 165         }
 166
 167         Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
 168         {:ok, activity}
 169
 170       error ->
 171         {:error, error}
 172     end
 173   end
 174
 175   defp get_participations({:ok, %{participations: participations}}), do: participations
 176   defp get_participations(_), do: []
 177
 178   def stream_out_participations(participations) do
 179     participations =
 180       participations
 181       |> Repo.preload(:user)
 182
 183     Enum.each(participations, fn participation ->
 184       Pleroma.Web.Streamer.stream("participation", participation)
 185     end)
 186   end
 187
 188   def stream_out(activity) do
 189     public = "https://www.w3.org/ns/activitystreams#Public"
 190
 191     if activity.data["type"] in ["Create", "Announce", "Delete"] do
 192       Pleroma.Web.Streamer.stream("user", activity)
 193       Pleroma.Web.Streamer.stream("list", activity)
 194
 195       if Enum.member?(activity.data["to"], public) do
 196         Pleroma.Web.Streamer.stream("public", activity)
 197
 198         if activity.local do
 199           Pleroma.Web.Streamer.stream("public:local", activity)
 200         end
 201
 202         if activity.data["type"] in ["Create"] do
 203           object = Object.normalize(activity)
 204
 205           object.data
 206           |> Map.get("tag", [])
 207           |> Enum.filter(fn tag -> is_bitstring(tag) end)
 208           |> Enum.each(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end)
 209
 210           if object.data["attachment"] != [] do
 211             Pleroma.Web.Streamer.stream("public:media", activity)
 212
 213             if activity.local do
 214               Pleroma.Web.Streamer.stream("public:local:media", activity)
 215             end
 216           end
 217         end
 218       else
 219         # TODO: Write test, replace with visibility test
 220         if !Enum.member?(activity.data["cc"] || [], public) &&
 221              !Enum.member?(
 222                activity.data["to"],
 223                User.get_cached_by_ap_id(activity.data["actor"]).follower_address
 224              ),
 225            do: Pleroma.Web.Streamer.stream("direct", activity)
 226       end
 227     end
 228   end
 229
 230   def create(%{to: to, actor: actor, context: context, object: object} = params, fake \\ false) do
 231     additional = params[:additional] || %{}
 232     # only accept false as false value
 233     local = !(params[:local] == false)
 234     published = params[:published]
 235
 236     with create_data <-
 237            make_create_data(
 238              %{to: to, actor: actor, published: published, context: context, object: object},
 239              additional
 240            ),
 241          {:ok, activity} <- insert(create_data, local, fake),
 242          {:fake, false, activity} <- {:fake, fake, activity},
 243          _ <- increase_replies_count_if_reply(create_data),
 244          # Changing note count prior to enqueuing federation task in order to avoid
 245          # race conditions on updating user.info
 246          {:ok, _actor} <- increase_note_count_if_public(actor, activity),
 247          :ok <- maybe_federate(activity) do
 248       {:ok, activity}
 249     else
 250       {:fake, true, activity} ->
 251         {:ok, activity}
 252     end
 253   end
 254
 255   def accept(%{to: to, actor: actor, object: object} = params) do
 256     # only accept false as false value
 257     local = !(params[:local] == false)
 258
 259     with data <- %{"to" => to, "type" => "Accept", "actor" => actor.ap_id, "object" => object},
 260          {:ok, activity} <- insert(data, local),
 261          :ok <- maybe_federate(activity) do
 262       {:ok, activity}
 263     end
 264   end
 265
 266   def reject(%{to: to, actor: actor, object: object} = params) do
 267     # only accept false as false value
 268     local = !(params[:local] == false)
 269
 270     with data <- %{"to" => to, "type" => "Reject", "actor" => actor.ap_id, "object" => object},
 271          {:ok, activity} <- insert(data, local),
 272          :ok <- maybe_federate(activity) do
 273       {:ok, activity}
 274     end
 275   end
 276
 277   def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
 278     # only accept false as false value
 279     local = !(params[:local] == false)
 280
 281     with data <- %{
 282            "to" => to,
 283            "cc" => cc,
 284            "type" => "Update",
 285            "actor" => actor,
 286            "object" => object
 287          },
 288          {:ok, activity} <- insert(data, local),
 289          :ok <- maybe_federate(activity) do
 290       {:ok, activity}
 291     end
 292   end
 293
 294   # TODO: This is weird, maybe we shouldn't check here if we can make the activity.
 295   def like(
 296         %User{ap_id: ap_id} = user,
 297         %Object{data: %{"id" => _}} = object,
 298         activity_id \\ nil,
 299         local \\ true
 300       ) do
 301     with nil <- get_existing_like(ap_id, object),
 302          like_data <- make_like_data(user, object, activity_id),
 303          {:ok, activity} <- insert(like_data, local),
 304          {:ok, object} <- add_like_to_object(activity, object),
 305          :ok <- maybe_federate(activity) do
 306       {:ok, activity, object}
 307     else
 308       %Activity{} = activity -> {:ok, activity, object}
 309       error -> {:error, error}
 310     end
 311   end
 312
 313   def unlike(
 314         %User{} = actor,
 315         %Object{} = object,
 316         activity_id \\ nil,
 317         local \\ true
 318       ) do
 319     with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
 320          unlike_data <- make_unlike_data(actor, like_activity, activity_id),
 321          {:ok, unlike_activity} <- insert(unlike_data, local),
 322          {:ok, _activity} <- Repo.delete(like_activity),
 323          {:ok, object} <- remove_like_from_object(like_activity, object),
 324          :ok <- maybe_federate(unlike_activity) do
 325       {:ok, unlike_activity, like_activity, object}
 326     else
 327       _e -> {:ok, object}
 328     end
 329   end
 330
 331   def announce(
 332         %User{ap_id: _} = user,
 333         %Object{data: %{"id" => _}} = object,
 334         activity_id \\ nil,
 335         local \\ true,
 336         public \\ true
 337       ) do
 338     with true <- is_public?(object),
 339          announce_data <- make_announce_data(user, object, activity_id, public),
 340          {:ok, activity} <- insert(announce_data, local),
 341          {:ok, object} <- add_announce_to_object(activity, object),
 342          :ok <- maybe_federate(activity) do
 343       {:ok, activity, object}
 344     else
 345       error -> {:error, error}
 346     end
 347   end
 348
 349   def unannounce(
 350         %User{} = actor,
 351         %Object{} = object,
 352         activity_id \\ nil,
 353         local \\ true
 354       ) do
 355     with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
 356          unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
 357          {:ok, unannounce_activity} <- insert(unannounce_data, local),
 358          :ok <- maybe_federate(unannounce_activity),
 359          {:ok, _activity} <- Repo.delete(announce_activity),
 360          {:ok, object} <- remove_announce_from_object(announce_activity, object) do
 361       {:ok, unannounce_activity, object}
 362     else
 363       _e -> {:ok, object}
 364     end
 365   end
 366
 367   def follow(follower, followed, activity_id \\ nil, local \\ true) do
 368     with data <- make_follow_data(follower, followed, activity_id),
 369          {:ok, activity} <- insert(data, local),
 370          :ok <- maybe_federate(activity) do
 371       {:ok, activity}
 372     end
 373   end
 374
 375   def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
 376     with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
 377          {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
 378          unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
 379          {:ok, activity} <- insert(unfollow_data, local),
 380          :ok <- maybe_federate(activity) do
 381       {:ok, activity}
 382     end
 383   end
 384
 385   def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, local \\ true) do
 386     user = User.get_cached_by_ap_id(actor)
 387     to = (object.data["to"] || []) ++ (object.data["cc"] || [])
 388
 389     with {:ok, object, activity} <- Object.delete(object),
 390          data <- %{
 391            "type" => "Delete",
 392            "actor" => actor,
 393            "object" => id,
 394            "to" => to,
 395            "deleted_activity_id" => activity && activity.id
 396          },
 397          {:ok, activity} <- insert(data, local),
 398          _ <- decrease_replies_count_if_reply(object),
 399          # Changing note count prior to enqueuing federation task in order to avoid
 400          # race conditions on updating user.info
 401          {:ok, _actor} <- decrease_note_count_if_public(user, object),
 402          :ok <- maybe_federate(activity) do
 403       {:ok, activity}
 404     end
 405   end
 406
 407   def block(blocker, blocked, activity_id \\ nil, local \\ true) do
 408     ap_config = Application.get_env(:pleroma, :activitypub)
 409     unfollow_blocked = Keyword.get(ap_config, :unfollow_blocked)
 410     outgoing_blocks = Keyword.get(ap_config, :outgoing_blocks)
 411
 412     with true <- unfollow_blocked do
 413       follow_activity = fetch_latest_follow(blocker, blocked)
 414
 415       if follow_activity do
 416         unfollow(blocker, blocked, nil, local)
 417       end
 418     end
 419
 420     with true <- outgoing_blocks,
 421          block_data <- make_block_data(blocker, blocked, activity_id),
 422          {:ok, activity} <- insert(block_data, local),
 423          :ok <- maybe_federate(activity) do
 424       {:ok, activity}
 425     else
 426       _e -> {:ok, nil}
 427     end
 428   end
 429
 430   def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
 431     with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
 432          unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
 433          {:ok, activity} <- insert(unblock_data, local),
 434          :ok <- maybe_federate(activity) do
 435       {:ok, activity}
 436     end
 437   end
 438
 439   def flag(
 440         %{
 441           actor: actor,
 442           context: context,
 443           account: account,
 444           statuses: statuses,
 445           content: content
 446         } = params
 447       ) do
 448     # only accept false as false value
 449     local = !(params[:local] == false)
 450     forward = !(params[:forward] == false)
 451
 452     additional = params[:additional] || %{}
 453
 454     params = %{
 455       actor: actor,
 456       context: context,
 457       account: account,
 458       statuses: statuses,
 459       content: content
 460     }
 461
 462     additional =
 463       if forward do
 464         Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
 465       else
 466         Map.merge(additional, %{"to" => [], "cc" => []})
 467       end
 468
 469     with flag_data <- make_flag_data(params, additional),
 470          {:ok, activity} <- insert(flag_data, local),
 471          :ok <- maybe_federate(activity) do
 472       Enum.each(User.all_superusers(), fn superuser ->
 473         superuser
 474         |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
 475         |> Pleroma.Emails.Mailer.deliver_async()
 476       end)
 477
 478       {:ok, activity}
 479     end
 480   end
 481
 482   defp fetch_activities_for_context_query(context, opts) do
 483     public = ["https://www.w3.org/ns/activitystreams#Public"]
 484
 485     recipients =
 486       if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
 487
 488     from(activity in Activity)
 489     |> restrict_blocked(opts)
 490     |> restrict_recipients(recipients, opts["user"])
 491     |> where(
 492       [activity],
 493       fragment(
 494         "?->>'type' = ? and ?->>'context' = ?",
 495         activity.data,
 496         "Create",
 497         activity.data,
 498         ^context
 499       )
 500     )
 501     |> order_by([activity], desc: activity.id)
 502   end
 503
 504   @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
 505   def fetch_activities_for_context(context, opts \\ %{}) do
 506     context
 507     |> fetch_activities_for_context_query(opts)
 508     |> Activity.with_preloaded_object()
 509     |> Repo.all()
 510   end
 511
 512   @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
 513           Pleroma.FlakeId.t() | nil
 514   def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
 515     context
 516     |> fetch_activities_for_context_query(opts)
 517     |> limit(1)
 518     |> select([a], a.id)
 519     |> Repo.one()
 520   end
 521
 522   def fetch_public_activities(opts \\ %{}) do
 523     q = fetch_activities_query(["https://www.w3.org/ns/activitystreams#Public"], opts)
 524
 525     q
 526     |> restrict_unlisted()
 527     |> Pagination.fetch_paginated(opts)
 528     |> Enum.reverse()
 529   end
 530
 531   @valid_visibilities ~w[direct unlisted public private]
 532
 533   defp restrict_visibility(query, %{visibility: visibility})
 534        when is_list(visibility) do
 535     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 536       query =
 537         from(
 538           a in query,
 539           where:
 540             fragment(
 541               "activity_visibility(?, ?, ?) = ANY (?)",
 542               a.actor,
 543               a.recipients,
 544               a.data,
 545               ^visibility
 546             )
 547         )
 548
 549       Ecto.Adapters.SQL.to_sql(:all, Repo, query)
 550
 551       query
 552     else
 553       Logger.error("Could not restrict visibility to #{visibility}")
 554     end
 555   end
 556
 557   defp restrict_visibility(query, %{visibility: visibility})
 558        when visibility in @valid_visibilities do
 559     query =
 560       from(
 561         a in query,
 562         where:
 563           fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 564       )
 565
 566     Ecto.Adapters.SQL.to_sql(:all, Repo, query)
 567
 568     query
 569   end
 570
 571   defp restrict_visibility(_query, %{visibility: visibility})
 572        when visibility not in @valid_visibilities do
 573     Logger.error("Could not restrict visibility to #{visibility}")
 574   end
 575
 576   defp restrict_visibility(query, _visibility), do: query
 577
 578   def fetch_user_activities(user, reading_user, params \\ %{}) do
 579     params =
 580       params
 581       |> Map.put("type", ["Create", "Announce"])
 582       |> Map.put("actor_id", user.ap_id)
 583       |> Map.put("whole_db", true)
 584       |> Map.put("pinned_activity_ids", user.info.pinned_activities)
 585
 586     recipients =
 587       if reading_user do
 588         ["https://www.w3.org/ns/activitystreams#Public"] ++
 589           [reading_user.ap_id | reading_user.following]
 590       else
 591         ["https://www.w3.org/ns/activitystreams#Public"]
 592       end
 593
 594     fetch_activities(recipients, params)
 595     |> Enum.reverse()
 596   end
 597
 598   defp restrict_since(query, %{"since_id" => ""}), do: query
 599
 600   defp restrict_since(query, %{"since_id" => since_id}) do
 601     from(activity in query, where: activity.id > ^since_id)
 602   end
 603
 604   defp restrict_since(query, _), do: query
 605
 606   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 607     raise "Can't use the child object without preloading!"
 608   end
 609
 610   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 611        when is_list(tag_reject) and tag_reject != [] do
 612     from(
 613       [_activity, object] in query,
 614       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 615     )
 616   end
 617
 618   defp restrict_tag_reject(query, _), do: query
 619
 620   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 621     raise "Can't use the child object without preloading!"
 622   end
 623
 624   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 625        when is_list(tag_all) and tag_all != [] do
 626     from(
 627       [_activity, object] in query,
 628       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 629     )
 630   end
 631
 632   defp restrict_tag_all(query, _), do: query
 633
 634   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 635     raise "Can't use the child object without preloading!"
 636   end
 637
 638   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 639     from(
 640       [_activity, object] in query,
 641       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 642     )
 643   end
 644
 645   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
 646     from(
 647       [_activity, object] in query,
 648       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
 649     )
 650   end
 651
 652   defp restrict_tag(query, _), do: query
 653
 654   defp restrict_to_cc(query, recipients_to, recipients_cc) do
 655     from(
 656       activity in query,
 657       where:
 658         fragment(
 659           "(?->'to' \\?| ?) or (?->'cc' \\?| ?)",
 660           activity.data,
 661           ^recipients_to,
 662           activity.data,
 663           ^recipients_cc
 664         )
 665     )
 666   end
 667
 668   defp restrict_recipients(query, [], _user), do: query
 669
 670   defp restrict_recipients(query, recipients, nil) do
 671     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 672   end
 673
 674   defp restrict_recipients(query, recipients, user) do
 675     from(
 676       activity in query,
 677       where: fragment("? && ?", ^recipients, activity.recipients),
 678       or_where: activity.actor == ^user.ap_id
 679     )
 680   end
 681
 682   defp restrict_local(query, %{"local_only" => true}) do
 683     from(activity in query, where: activity.local == true)
 684   end
 685
 686   defp restrict_local(query, _), do: query
 687
 688   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 689     from(activity in query, where: activity.actor == ^actor_id)
 690   end
 691
 692   defp restrict_actor(query, _), do: query
 693
 694   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 695     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 696   end
 697
 698   defp restrict_type(query, %{"type" => type}) do
 699     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 700   end
 701
 702   defp restrict_type(query, _), do: query
 703
 704   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 705     from(
 706       activity in query,
 707       where: fragment(~s(? <@ (? #> '{"object","likes"}'\)), ^ap_id, activity.data)
 708     )
 709   end
 710
 711   defp restrict_favorited_by(query, _), do: query
 712
 713   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
 714     raise "Can't use the child object without preloading!"
 715   end
 716
 717   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 718     from(
 719       [_activity, object] in query,
 720       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 721     )
 722   end
 723
 724   defp restrict_media(query, _), do: query
 725
 726   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
 727     from(
 728       activity in query,
 729       where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
 730     )
 731   end
 732
 733   defp restrict_replies(query, _), do: query
 734
 735   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
 736     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 737   end
 738
 739   defp restrict_reblogs(query, _), do: query
 740
 741   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 742
 743   defp restrict_muted(query, %{"muting_user" => %User{info: info}}) do
 744     mutes = info.mutes
 745
 746     from(
 747       activity in query,
 748       where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 749       where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 750     )
 751   end
 752
 753   defp restrict_muted(query, _), do: query
 754
 755   defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
 756     blocks = info.blocks || []
 757     domain_blocks = info.domain_blocks || []
 758
 759     from(
 760       activity in query,
 761       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 762       where: fragment("not (? && ?)", activity.recipients, ^blocks),
 763       where:
 764         fragment(
 765           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 766           activity.data,
 767           activity.data,
 768           ^blocks
 769         ),
 770       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks)
 771     )
 772   end
 773
 774   defp restrict_blocked(query, _), do: query
 775
 776   defp restrict_unlisted(query) do
 777     from(
 778       activity in query,
 779       where:
 780         fragment(
 781           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 782           activity.data,
 783           ^["https://www.w3.org/ns/activitystreams#Public"]
 784         )
 785     )
 786   end
 787
 788   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 789     from(activity in query, where: activity.id in ^ids)
 790   end
 791
 792   defp restrict_pinned(query, _), do: query
 793
 794   defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
 795     muted_reblogs = info.muted_reblogs || []
 796
 797     from(
 798       activity in query,
 799       where:
 800         fragment(
 801           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 802           activity.data,
 803           activity.actor,
 804           ^muted_reblogs
 805         )
 806     )
 807   end
 808
 809   defp restrict_muted_reblogs(query, _), do: query
 810
 811   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 812
 813   defp maybe_preload_objects(query, _) do
 814     query
 815     |> Activity.with_preloaded_object()
 816   end
 817
 818   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
 819
 820   defp maybe_preload_bookmarks(query, opts) do
 821     query
 822     |> Activity.with_preloaded_bookmark(opts["user"])
 823   end
 824
 825   defp maybe_order(query, %{order: :desc}) do
 826     query
 827     |> order_by(desc: :id)
 828   end
 829
 830   defp maybe_order(query, %{order: :asc}) do
 831     query
 832     |> order_by(asc: :id)
 833   end
 834
 835   defp maybe_order(query, _), do: query
 836
 837   def fetch_activities_query(recipients, opts \\ %{}) do
 838     base_query = from(activity in Activity)
 839
 840     base_query
 841     |> maybe_preload_objects(opts)
 842     |> maybe_preload_bookmarks(opts)
 843     |> maybe_order(opts)
 844     |> restrict_recipients(recipients, opts["user"])
 845     |> restrict_tag(opts)
 846     |> restrict_tag_reject(opts)
 847     |> restrict_tag_all(opts)
 848     |> restrict_since(opts)
 849     |> restrict_local(opts)
 850     |> restrict_actor(opts)
 851     |> restrict_type(opts)
 852     |> restrict_favorited_by(opts)
 853     |> restrict_blocked(opts)
 854     |> restrict_muted(opts)
 855     |> restrict_media(opts)
 856     |> restrict_visibility(opts)
 857     |> restrict_replies(opts)
 858     |> restrict_reblogs(opts)
 859     |> restrict_pinned(opts)
 860     |> restrict_muted_reblogs(opts)
 861   end
 862
 863   def fetch_activities(recipients, opts \\ %{}) do
 864     fetch_activities_query(recipients, opts)
 865     |> Pagination.fetch_paginated(opts)
 866     |> Enum.reverse()
 867   end
 868
 869   def fetch_activities_bounded(recipients_to, recipients_cc, opts \\ %{}) do
 870     fetch_activities_query([], opts)
 871     |> restrict_to_cc(recipients_to, recipients_cc)
 872     |> Pagination.fetch_paginated(opts)
 873     |> Enum.reverse()
 874   end
 875
 876   def upload(file, opts \\ []) do
 877     with {:ok, data} <- Upload.store(file, opts) do
 878       obj_data =
 879         if opts[:actor] do
 880           Map.put(data, "actor", opts[:actor])
 881         else
 882           data
 883         end
 884
 885       Repo.insert(%Object{data: obj_data})
 886     end
 887   end
 888
 889   def user_data_from_user_object(data) do
 890     avatar =
 891       data["icon"]["url"] &&
 892         %{
 893           "type" => "Image",
 894           "url" => [%{"href" => data["icon"]["url"]}]
 895         }
 896
 897     banner =
 898       data["image"]["url"] &&
 899         %{
 900           "type" => "Image",
 901           "url" => [%{"href" => data["image"]["url"]}]
 902         }
 903
 904     locked = data["manuallyApprovesFollowers"] || false
 905     data = Transmogrifier.maybe_fix_user_object(data)
 906
 907     user_data = %{
 908       ap_id: data["id"],
 909       info: %{
 910         "ap_enabled" => true,
 911         "source_data" => data,
 912         "banner" => banner,
 913         "locked" => locked
 914       },
 915       avatar: avatar,
 916       name: data["name"],
 917       follower_address: data["followers"],
 918       bio: data["summary"]
 919     }
 920
 921     # nickname can be nil because of virtual actors
 922     user_data =
 923       if data["preferredUsername"] do
 924         Map.put(
 925           user_data,
 926           :nickname,
 927           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
 928         )
 929       else
 930         Map.put(user_data, :nickname, nil)
 931       end
 932
 933     {:ok, user_data}
 934   end
 935
 936   def fetch_and_prepare_user_from_ap_id(ap_id) do
 937     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id) do
 938       user_data_from_user_object(data)
 939     else
 940       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
 941     end
 942   end
 943
 944   def make_user_from_ap_id(ap_id) do
 945     if _user = User.get_cached_by_ap_id(ap_id) do
 946       Transmogrifier.upgrade_user_from_ap_id(ap_id)
 947     else
 948       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
 949         User.insert_or_update_user(data)
 950       else
 951         e -> {:error, e}
 952       end
 953     end
 954   end
 955
 956   def make_user_from_nickname(nickname) do
 957     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
 958       make_user_from_ap_id(ap_id)
 959     else
 960       _e -> {:error, "No AP id in WebFinger"}
 961     end
 962   end
 963
 964   def should_federate?(inbox, public) do
 965     if public do
 966       true
 967     else
 968       inbox_info = URI.parse(inbox)
 969       !Enum.member?(Pleroma.Config.get([:instance, :quarantined_instances], []), inbox_info.host)
 970     end
 971   end
 972
 973   def publish(actor, activity) do
 974     remote_followers =
 975       if actor.follower_address in activity.recipients do
 976         {:ok, followers} = User.get_followers(actor)
 977         followers |> Enum.filter(&(!&1.local))
 978       else
 979         []
 980       end
 981
 982     public = is_public?(activity)
 983
 984     {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
 985     json = Jason.encode!(data)
 986
 987     (Pleroma.Web.Salmon.remote_users(activity) ++ remote_followers)
 988     |> Enum.filter(fn user -> User.ap_enabled?(user) end)
 989     |> Enum.map(fn %{info: %{source_data: data}} ->
 990       (is_map(data["endpoints"]) && Map.get(data["endpoints"], "sharedInbox")) || data["inbox"]
 991     end)
 992     |> Enum.uniq()
 993     |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
 994     |> Instances.filter_reachable()
 995     |> Enum.each(fn {inbox, unreachable_since} ->
 996       Federator.publish_single_ap(%{
 997         inbox: inbox,
 998         json: json,
 999         actor: actor,
1000         id: activity.data["id"],
1001         unreachable_since: unreachable_since
1002       })
1003     end)
1004   end
1005
1006   def publish_one(%{inbox: inbox, json: json, actor: actor, id: id} = params) do
1007     Logger.info("Federating #{id} to #{inbox}")
1008     host = URI.parse(inbox).host
1009
1010     digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64())
1011
1012     date =
1013       NaiveDateTime.utc_now()
1014       |> Timex.format!("{WDshort}, {0D} {Mshort} {YYYY} {h24}:{m}:{s} GMT")
1015
1016     signature =
1017       Pleroma.Web.HTTPSignatures.sign(actor, %{
1018         host: host,
1019         "content-length": byte_size(json),
1020         digest: digest,
1021         date: date
1022       })
1023
1024     with {:ok, %{status: code}} when code in 200..299 <-
1025            result =
1026              @httpoison.post(
1027                inbox,
1028                json,
1029                [
1030                  {"Content-Type", "application/activity+json"},
1031                  {"Date", date},
1032                  {"signature", signature},
1033                  {"digest", digest}
1034                ]
1035              ) do
1036       if !Map.has_key?(params, :unreachable_since) || params[:unreachable_since],
1037         do: Instances.set_reachable(inbox)
1038
1039       result
1040     else
1041       {_post_result, response} ->
1042         unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
1043         {:error, response}
1044     end
1045   end
1046
1047   # filter out broken threads
1048   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1049     entire_thread_visible_for_user?(activity, user)
1050   end
1051
1052   # do post-processing on a specific activity
1053   def contain_activity(%Activity{} = activity, %User{} = user) do
1054     contain_broken_threads(activity, user)
1055   end
1056
1057   # do post-processing on a timeline
1058   def contain_timeline(timeline, user) do
1059     timeline
1060     |> Enum.filter(fn activity ->
1061       contain_activity(activity, user)
1062     end)
1063   end
1064 end