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

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