git.squeep.com Git - akkoma/blob - lib/pleroma/web/activity_pub/activity_pub.ex

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