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   def fetch_activities_query(recipients, opts \\ %{}) do
 819     base_query = from(activity in Activity)
 820
 821     base_query
 822     |> maybe_preload_objects(opts)
 823     |> restrict_recipients(recipients, opts["user"])
 824     |> restrict_tag(opts)
 825     |> restrict_tag_reject(opts)
 826     |> restrict_tag_all(opts)
 827     |> restrict_since(opts)
 828     |> restrict_local(opts)
 829     |> restrict_actor(opts)
 830     |> restrict_type(opts)
 831     |> restrict_favorited_by(opts)
 832     |> restrict_blocked(opts)
 833     |> restrict_muted(opts)
 834     |> restrict_media(opts)
 835     |> restrict_visibility(opts)
 836     |> restrict_replies(opts)
 837     |> restrict_reblogs(opts)
 838     |> restrict_pinned(opts)
 839     |> restrict_muted_reblogs(opts)
 840   end
 841
 842   def fetch_activities(recipients, opts \\ %{}) do
 843     fetch_activities_query(recipients, opts)
 844     |> Pagination.fetch_paginated(opts)
 845     |> Enum.reverse()
 846   end
 847
 848   def fetch_activities_bounded(recipients_to, recipients_cc, opts \\ %{}) do
 849     fetch_activities_query([], opts)
 850     |> restrict_to_cc(recipients_to, recipients_cc)
 851     |> Pagination.fetch_paginated(opts)
 852     |> Enum.reverse()
 853   end
 854
 855   def upload(file, opts \\ []) do
 856     with {:ok, data} <- Upload.store(file, opts) do
 857       obj_data =
 858         if opts[:actor] do
 859           Map.put(data, "actor", opts[:actor])
 860         else
 861           data
 862         end
 863
 864       Repo.insert(%Object{data: obj_data})
 865     end
 866   end
 867
 868   def user_data_from_user_object(data) do
 869     avatar =
 870       data["icon"]["url"] &&
 871         %{
 872           "type" => "Image",
 873           "url" => [%{"href" => data["icon"]["url"]}]
 874         }
 875
 876     banner =
 877       data["image"]["url"] &&
 878         %{
 879           "type" => "Image",
 880           "url" => [%{"href" => data["image"]["url"]}]
 881         }
 882
 883     locked = data["manuallyApprovesFollowers"] || false
 884     data = Transmogrifier.maybe_fix_user_object(data)
 885
 886     user_data = %{
 887       ap_id: data["id"],
 888       info: %{
 889         "ap_enabled" => true,
 890         "source_data" => data,
 891         "banner" => banner,
 892         "locked" => locked
 893       },
 894       avatar: avatar,
 895       name: data["name"],
 896       follower_address: data["followers"],
 897       bio: data["summary"]
 898     }
 899
 900     # nickname can be nil because of virtual actors
 901     user_data =
 902       if data["preferredUsername"] do
 903         Map.put(
 904           user_data,
 905           :nickname,
 906           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
 907         )
 908       else
 909         Map.put(user_data, :nickname, nil)
 910       end
 911
 912     {:ok, user_data}
 913   end
 914
 915   def fetch_and_prepare_user_from_ap_id(ap_id) do
 916     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id) do
 917       user_data_from_user_object(data)
 918     else
 919       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
 920     end
 921   end
 922
 923   def make_user_from_ap_id(ap_id) do
 924     if _user = User.get_cached_by_ap_id(ap_id) do
 925       Transmogrifier.upgrade_user_from_ap_id(ap_id)
 926     else
 927       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
 928         User.insert_or_update_user(data)
 929       else
 930         e -> {:error, e}
 931       end
 932     end
 933   end
 934
 935   def make_user_from_nickname(nickname) do
 936     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
 937       make_user_from_ap_id(ap_id)
 938     else
 939       _e -> {:error, "No AP id in WebFinger"}
 940     end
 941   end
 942
 943   def should_federate?(inbox, public) do
 944     if public do
 945       true
 946     else
 947       inbox_info = URI.parse(inbox)
 948       !Enum.member?(Pleroma.Config.get([:instance, :quarantined_instances], []), inbox_info.host)
 949     end
 950   end
 951
 952   def publish(actor, activity) do
 953     remote_followers =
 954       if actor.follower_address in activity.recipients do
 955         {:ok, followers} = User.get_followers(actor)
 956         followers |> Enum.filter(&(!&1.local))
 957       else
 958         []
 959       end
 960
 961     public = is_public?(activity)
 962
 963     {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
 964     json = Jason.encode!(data)
 965
 966     (Pleroma.Web.Salmon.remote_users(activity) ++ remote_followers)
 967     |> Enum.filter(fn user -> User.ap_enabled?(user) end)
 968     |> Enum.map(fn %{info: %{source_data: data}} ->
 969       (is_map(data["endpoints"]) && Map.get(data["endpoints"], "sharedInbox")) || data["inbox"]
 970     end)
 971     |> Enum.uniq()
 972     |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
 973     |> Instances.filter_reachable()
 974     |> Enum.each(fn {inbox, unreachable_since} ->
 975       Federator.publish_single_ap(%{
 976         inbox: inbox,
 977         json: json,
 978         actor: actor,
 979         id: activity.data["id"],
 980         unreachable_since: unreachable_since
 981       })
 982     end)
 983   end
 984
 985   def publish_one(%{inbox: inbox, json: json, actor: actor, id: id} = params) do
 986     Logger.info("Federating #{id} to #{inbox}")
 987     host = URI.parse(inbox).host
 988
 989     digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64())
 990
 991     date =
 992       NaiveDateTime.utc_now()
 993       |> Timex.format!("{WDshort}, {0D} {Mshort} {YYYY} {h24}:{m}:{s} GMT")
 994
 995     signature =
 996       Pleroma.Web.HTTPSignatures.sign(actor, %{
 997         host: host,
 998         "content-length": byte_size(json),
 999         digest: digest,
1000         date: date
1001       })
1002
1003     with {:ok, %{status: code}} when code in 200..299 <-
1004            result =
1005              @httpoison.post(
1006                inbox,
1007                json,
1008                [
1009                  {"Content-Type", "application/activity+json"},
1010                  {"Date", date},
1011                  {"signature", signature},
1012                  {"digest", digest}
1013                ]
1014              ) do
1015       if !Map.has_key?(params, :unreachable_since) || params[:unreachable_since],
1016         do: Instances.set_reachable(inbox)
1017
1018       result
1019     else
1020       {_post_result, response} ->
1021         unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
1022         {:error, response}
1023     end
1024   end
1025
1026   # filter out broken threads
1027   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1028     entire_thread_visible_for_user?(activity, user)
1029   end
1030
1031   # do post-processing on a specific activity
1032   def contain_activity(%Activity{} = activity, %User{} = user) do
1033     contain_broken_threads(activity, user)
1034   end
1035
1036   # do post-processing on a timeline
1037   def contain_timeline(timeline, user) do
1038     timeline
1039     |> Enum.filter(fn activity ->
1040       contain_activity(activity, user)
1041     end)
1042   end
1043 end