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