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