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