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