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 (? && ?)", activity.recipients, ^blocks),
 716       where:
 717         fragment(
 718           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 719           activity.data,
 720           activity.data,
 721           ^blocks
 722         ),
 723       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks)
 724     )
 725   end
 726
 727   defp restrict_blocked(query, _), do: query
 728
 729   defp restrict_unlisted(query) do
 730     from(
 731       activity in query,
 732       where:
 733         fragment(
 734           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 735           activity.data,
 736           ^["https://www.w3.org/ns/activitystreams#Public"]
 737         )
 738     )
 739   end
 740
 741   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 742     from(activity in query, where: activity.id in ^ids)
 743   end
 744
 745   defp restrict_pinned(query, _), do: query
 746
 747   defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
 748     muted_reblogs = info.muted_reblogs || []
 749
 750     from(
 751       activity in query,
 752       where:
 753         fragment(
 754           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 755           activity.data,
 756           activity.actor,
 757           ^muted_reblogs
 758         )
 759     )
 760   end
 761
 762   defp restrict_muted_reblogs(query, _), do: query
 763
 764   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 765
 766   defp maybe_preload_objects(query, _) do
 767     query
 768     |> Activity.with_preloaded_object()
 769   end
 770
 771   def fetch_activities_query(recipients, opts \\ %{}) do
 772     base_query = from(activity in Activity)
 773
 774     base_query
 775     |> maybe_preload_objects(opts)
 776     |> restrict_recipients(recipients, opts["user"])
 777     |> restrict_tag(opts)
 778     |> restrict_tag_reject(opts)
 779     |> restrict_tag_all(opts)
 780     |> restrict_since(opts)
 781     |> restrict_local(opts)
 782     |> restrict_actor(opts)
 783     |> restrict_type(opts)
 784     |> restrict_favorited_by(opts)
 785     |> restrict_blocked(opts)
 786     |> restrict_muted(opts)
 787     |> restrict_media(opts)
 788     |> restrict_visibility(opts)
 789     |> restrict_replies(opts)
 790     |> restrict_reblogs(opts)
 791     |> restrict_pinned(opts)
 792     |> restrict_muted_reblogs(opts)
 793   end
 794
 795   def fetch_activities(recipients, opts \\ %{}) do
 796     fetch_activities_query(recipients, opts)
 797     |> Pagination.fetch_paginated(opts)
 798     |> Enum.reverse()
 799   end
 800
 801   def fetch_activities_bounded(recipients_to, recipients_cc, opts \\ %{}) do
 802     fetch_activities_query([], opts)
 803     |> restrict_to_cc(recipients_to, recipients_cc)
 804     |> Pagination.fetch_paginated(opts)
 805     |> Enum.reverse()
 806   end
 807
 808   def upload(file, opts \\ []) do
 809     with {:ok, data} <- Upload.store(file, opts) do
 810       obj_data =
 811         if opts[:actor] do
 812           Map.put(data, "actor", opts[:actor])
 813         else
 814           data
 815         end
 816
 817       Repo.insert(%Object{data: obj_data})
 818     end
 819   end
 820
 821   def user_data_from_user_object(data) do
 822     avatar =
 823       data["icon"]["url"] &&
 824         %{
 825           "type" => "Image",
 826           "url" => [%{"href" => data["icon"]["url"]}]
 827         }
 828
 829     banner =
 830       data["image"]["url"] &&
 831         %{
 832           "type" => "Image",
 833           "url" => [%{"href" => data["image"]["url"]}]
 834         }
 835
 836     locked = data["manuallyApprovesFollowers"] || false
 837     data = Transmogrifier.maybe_fix_user_object(data)
 838
 839     user_data = %{
 840       ap_id: data["id"],
 841       info: %{
 842         "ap_enabled" => true,
 843         "source_data" => data,
 844         "banner" => banner,
 845         "locked" => locked
 846       },
 847       avatar: avatar,
 848       name: data["name"],
 849       follower_address: data["followers"],
 850       bio: data["summary"]
 851     }
 852
 853     # nickname can be nil because of virtual actors
 854     user_data =
 855       if data["preferredUsername"] do
 856         Map.put(
 857           user_data,
 858           :nickname,
 859           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
 860         )
 861       else
 862         Map.put(user_data, :nickname, nil)
 863       end
 864
 865     {:ok, user_data}
 866   end
 867
 868   def fetch_and_prepare_user_from_ap_id(ap_id) do
 869     with {:ok, data} <- fetch_and_contain_remote_object_from_id(ap_id) do
 870       user_data_from_user_object(data)
 871     else
 872       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
 873     end
 874   end
 875
 876   def make_user_from_ap_id(ap_id) do
 877     if _user = User.get_by_ap_id(ap_id) do
 878       Transmogrifier.upgrade_user_from_ap_id(ap_id)
 879     else
 880       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
 881         User.insert_or_update_user(data)
 882       else
 883         e -> {:error, e}
 884       end
 885     end
 886   end
 887
 888   def make_user_from_nickname(nickname) do
 889     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
 890       make_user_from_ap_id(ap_id)
 891     else
 892       _e -> {:error, "No AP id in WebFinger"}
 893     end
 894   end
 895
 896   def should_federate?(inbox, public) do
 897     if public do
 898       true
 899     else
 900       inbox_info = URI.parse(inbox)
 901       !Enum.member?(Pleroma.Config.get([:instance, :quarantined_instances], []), inbox_info.host)
 902     end
 903   end
 904
 905   def publish(actor, activity) do
 906     remote_followers =
 907       if actor.follower_address in activity.recipients do
 908         {:ok, followers} = User.get_followers(actor)
 909         followers |> Enum.filter(&(!&1.local))
 910       else
 911         []
 912       end
 913
 914     public = is_public?(activity)
 915
 916     {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
 917     json = Jason.encode!(data)
 918
 919     (Pleroma.Web.Salmon.remote_users(activity) ++ remote_followers)
 920     |> Enum.filter(fn user -> User.ap_enabled?(user) end)
 921     |> Enum.map(fn %{info: %{source_data: data}} ->
 922       (is_map(data["endpoints"]) && Map.get(data["endpoints"], "sharedInbox")) || data["inbox"]
 923     end)
 924     |> Enum.uniq()
 925     |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
 926     |> Instances.filter_reachable()
 927     |> Enum.each(fn {inbox, unreachable_since} ->
 928       Federator.publish_single_ap(%{
 929         inbox: inbox,
 930         json: json,
 931         actor: actor,
 932         id: activity.data["id"],
 933         unreachable_since: unreachable_since
 934       })
 935     end)
 936   end
 937
 938   def publish_one(%{inbox: inbox, json: json, actor: actor, id: id} = params) do
 939     Logger.info("Federating #{id} to #{inbox}")
 940     host = URI.parse(inbox).host
 941
 942     digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64())
 943
 944     date =
 945       NaiveDateTime.utc_now()
 946       |> Timex.format!("{WDshort}, {0D} {Mshort} {YYYY} {h24}:{m}:{s} GMT")
 947
 948     signature =
 949       Pleroma.Web.HTTPSignatures.sign(actor, %{
 950         host: host,
 951         "content-length": byte_size(json),
 952         digest: digest,
 953         date: date
 954       })
 955
 956     with {:ok, %{status: code}} when code in 200..299 <-
 957            result =
 958              @httpoison.post(
 959                inbox,
 960                json,
 961                [
 962                  {"Content-Type", "application/activity+json"},
 963                  {"Date", date},
 964                  {"signature", signature},
 965                  {"digest", digest}
 966                ]
 967              ) do
 968       if !Map.has_key?(params, :unreachable_since) || params[:unreachable_since],
 969         do: Instances.set_reachable(inbox)
 970
 971       result
 972     else
 973       {_post_result, response} ->
 974         unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
 975         {:error, response}
 976     end
 977   end
 978
 979   # TODO:
 980   # This will create a Create activity, which we need internally at the moment.
 981   def fetch_object_from_id(id) do
 982     if object = Object.get_cached_by_ap_id(id) do
 983       {:ok, object}
 984     else
 985       with {:ok, data} <- fetch_and_contain_remote_object_from_id(id),
 986            nil <- Object.normalize(data),
 987            params <- %{
 988              "type" => "Create",
 989              "to" => data["to"],
 990              "cc" => data["cc"],
 991              "actor" => data["actor"] || data["attributedTo"],
 992              "object" => data
 993            },
 994            :ok <- Transmogrifier.contain_origin(id, params),
 995            {:ok, activity} <- Transmogrifier.handle_incoming(params) do
 996         {:ok, Object.normalize(activity)}
 997       else
 998         {:error, {:reject, nil}} ->
 999           {:reject, nil}
1000
1001         object = %Object{} ->
1002           {:ok, object}
1003
1004         _e ->
1005           Logger.info("Couldn't get object via AP, trying out OStatus fetching...")
1006
1007           case OStatus.fetch_activity_from_url(id) do
1008             {:ok, [activity | _]} -> {:ok, Object.normalize(activity)}
1009             e -> e
1010           end
1011       end
1012     end
1013   end
1014
1015   def fetch_and_contain_remote_object_from_id(id) do
1016     Logger.info("Fetching object #{id} via AP")
1017
1018     with true <- String.starts_with?(id, "http"),
1019          {:ok, %{body: body, status: code}} when code in 200..299 <-
1020            @httpoison.get(
1021              id,
1022              [{:Accept, "application/activity+json"}]
1023            ),
1024          {:ok, data} <- Jason.decode(body),
1025          :ok <- Transmogrifier.contain_origin_from_id(id, data) do
1026       {:ok, data}
1027     else
1028       e ->
1029         {:error, e}
1030     end
1031   end
1032
1033   # filter out broken threads
1034   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1035     entire_thread_visible_for_user?(activity, user)
1036   end
1037
1038   # do post-processing on a specific activity
1039   def contain_activity(%Activity{} = activity, %User{} = user) do
1040     contain_broken_threads(activity, user)
1041   end
1042
1043   # do post-processing on a timeline
1044   def contain_timeline(timeline, user) do
1045     timeline
1046     |> Enum.filter(fn activity ->
1047       contain_activity(activity, user)
1048     end)
1049   end
1050 end