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