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

   1 defmodule Pleroma.Web.ActivityPub.ActivityPub do
   2   alias Pleroma.{Activity, Repo, Object, Upload, User, Notification}
   3   alias Pleroma.Web.ActivityPub.Transmogrifier
   4   alias Pleroma.Web.WebFinger
   5   alias Pleroma.Web.Federator
   6   alias Pleroma.Web.OStatus
   7   import Ecto.Query
   8   import Pleroma.Web.ActivityPub.Utils
   9   require Logger
  10
  11   @httpoison Application.get_env(:pleroma, :httpoison)
  12
  13   @instance Application.get_env(:pleroma, :instance)
  14   @rewrite_policy Keyword.get(@instance, :rewrite_policy)
  15
  16   def get_recipients(data) do
  17     (data["to"] || []) ++ (data["cc"] || [])
  18   end
  19
  20   def insert(map, local \\ true) when is_map(map) do
  21     with nil <- Activity.get_by_ap_id(map["id"]),
  22          map <- lazy_put_activity_defaults(map),
  23          {:ok, map} <- @rewrite_policy.filter(map),
  24          :ok <- insert_full_object(map) do
  25       {:ok, activity} =
  26         Repo.insert(%Activity{
  27           data: map,
  28           local: local,
  29           actor: map["actor"],
  30           recipients: get_recipients(map)
  31         })
  32
  33       Notification.create_notifications(activity)
  34       stream_out(activity)
  35       {:ok, activity}
  36     else
  37       %Activity{} = activity -> {:ok, activity}
  38       error -> {:error, error}
  39     end
  40   end
  41
  42   def stream_out(activity) do
  43     public = "https://www.w3.org/ns/activitystreams#Public"
  44
  45     if activity.data["type"] in ["Create", "Announce"] do
  46       Pleroma.Web.Streamer.stream("user", activity)
  47
  48       if Enum.member?(activity.data["to"], public) do
  49         Pleroma.Web.Streamer.stream("public", activity)
  50
  51         if activity.local do
  52           Pleroma.Web.Streamer.stream("public:local", activity)
  53         end
  54       else
  55         if !Enum.member?(activity.data["cc"] || [], public) &&
  56              !Enum.member?(
  57                activity.data["to"],
  58                User.get_by_ap_id(activity.data["actor"]).follower_address
  59              ), do: Pleroma.Web.Streamer.stream("direct", activity)
  60       end
  61     end
  62   end
  63
  64   def create(%{to: to, actor: actor, context: context, object: object} = params) do
  65     additional = params[:additional] || %{}
  66     # only accept false as false value
  67     local = !(params[:local] == false)
  68     published = params[:published]
  69
  70     with create_data <-
  71            make_create_data(
  72              %{to: to, actor: actor, published: published, context: context, object: object},
  73              additional
  74            ),
  75          {:ok, activity} <- insert(create_data, local),
  76          :ok <- maybe_federate(activity),
  77          {:ok, _actor} <- User.increase_note_count(actor) do
  78       {:ok, activity}
  79     end
  80   end
  81
  82   def accept(%{to: to, actor: actor, object: object} = params) do
  83     # only accept false as false value
  84     local = !(params[:local] == false)
  85
  86     with data <- %{"to" => to, "type" => "Accept", "actor" => actor, "object" => object},
  87          {:ok, activity} <- insert(data, local),
  88          :ok <- maybe_federate(activity) do
  89       {:ok, activity}
  90     end
  91   end
  92
  93   def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
  94     # only accept false as false value
  95     local = !(params[:local] == false)
  96
  97     with data <- %{
  98            "to" => to,
  99            "cc" => cc,
 100            "type" => "Update",
 101            "actor" => actor,
 102            "object" => object
 103          },
 104          {:ok, activity} <- insert(data, local),
 105          :ok <- maybe_federate(activity) do
 106       {:ok, activity}
 107     end
 108   end
 109
 110   # TODO: This is weird, maybe we shouldn't check here if we can make the activity.
 111   def like(
 112         %User{ap_id: ap_id} = user,
 113         %Object{data: %{"id" => _}} = object,
 114         activity_id \\ nil,
 115         local \\ true
 116       ) do
 117     with nil <- get_existing_like(ap_id, object),
 118          like_data <- make_like_data(user, object, activity_id),
 119          {:ok, activity} <- insert(like_data, local),
 120          {:ok, object} <- add_like_to_object(activity, object),
 121          :ok <- maybe_federate(activity) do
 122       {:ok, activity, object}
 123     else
 124       %Activity{} = activity -> {:ok, activity, object}
 125       error -> {:error, error}
 126     end
 127   end
 128
 129   def unlike(%User{} = actor, %Object{} = object) do
 130     with %Activity{} = activity <- get_existing_like(actor.ap_id, object),
 131          {:ok, _activity} <- Repo.delete(activity),
 132          {:ok, object} <- remove_like_from_object(activity, object) do
 133       {:ok, object}
 134     else
 135       _e -> {:ok, object}
 136     end
 137   end
 138
 139   def announce(
 140         %User{ap_id: _} = user,
 141         %Object{data: %{"id" => _}} = object,
 142         activity_id \\ nil,
 143         local \\ true
 144       ) do
 145     with true <- is_public?(object),
 146          announce_data <- make_announce_data(user, object, activity_id),
 147          {:ok, activity} <- insert(announce_data, local),
 148          {:ok, object} <- add_announce_to_object(activity, object),
 149          :ok <- maybe_federate(activity) do
 150       {:ok, activity, object}
 151     else
 152       error -> {:error, error}
 153     end
 154   end
 155
 156   def follow(follower, followed, activity_id \\ nil, local \\ true) do
 157     with data <- make_follow_data(follower, followed, activity_id),
 158          {:ok, activity} <- insert(data, local),
 159          :ok <- maybe_federate(activity) do
 160       {:ok, activity}
 161     end
 162   end
 163
 164   def unfollow(follower, followed, local \\ true) do
 165     with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
 166          unfollow_data <- make_unfollow_data(follower, followed, follow_activity),
 167          {:ok, activity} <- insert(unfollow_data, local),
 168          :ok,
 169          maybe_federate(activity) do
 170       {:ok, activity}
 171     end
 172   end
 173
 174   def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, local \\ true) do
 175     user = User.get_cached_by_ap_id(actor)
 176
 177     data = %{
 178       "type" => "Delete",
 179       "actor" => actor,
 180       "object" => id,
 181       "to" => [user.follower_address, "https://www.w3.org/ns/activitystreams#Public"]
 182     }
 183
 184     with Repo.delete(object),
 185          Repo.delete_all(Activity.all_non_create_by_object_ap_id_q(id)),
 186          {:ok, activity} <- insert(data, local),
 187          :ok <- maybe_federate(activity),
 188          {:ok, _actor} <- User.decrease_note_count(user) do
 189       {:ok, activity}
 190     end
 191   end
 192
 193   def fetch_activities_for_context(context, opts \\ %{}) do
 194     public = ["https://www.w3.org/ns/activitystreams#Public"]
 195
 196     recipients =
 197       if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
 198
 199     query = from(activity in Activity)
 200
 201     query =
 202       query
 203       |> restrict_blocked(opts)
 204       |> restrict_recipients(recipients, opts["user"])
 205
 206     query =
 207       from(
 208         activity in query,
 209         where:
 210           fragment(
 211             "?->>'type' = ? and ?->>'context' = ?",
 212             activity.data,
 213             "Create",
 214             activity.data,
 215             ^context
 216           ),
 217         order_by: [desc: :id]
 218       )
 219
 220     Repo.all(query)
 221   end
 222
 223   # TODO: Make this work properly with unlisted.
 224   def fetch_public_activities(opts \\ %{}) do
 225     q = fetch_activities_query(["https://www.w3.org/ns/activitystreams#Public"], opts)
 226
 227     q
 228     |> Repo.all()
 229     |> Enum.reverse()
 230   end
 231
 232   @valid_visibilities ~w[direct unlisted public private]
 233
 234   defp restrict_visibility(query, %{visibility: "direct"}) do
 235     public = "https://www.w3.org/ns/activitystreams#Public"
 236
 237     from(
 238       activity in query,
 239       join: sender in User,
 240       on: sender.ap_id == activity.actor,
 241       where:
 242         fragment("not data->'to' \\? ?", ^public) and fragment("not data->'cc' \\? ?", ^public) and
 243           fragment("not data->'to' \\? ?", sender.follower_address)
 244     )
 245   end
 246
 247   defp restrict_visibility(_query, %{visibility: visibility})
 248        when visibility not in @valid_visibilities do
 249     Logger.error("Could not restrict visibility to #{visibility}")
 250   end
 251
 252   defp restrict_visibility(query, _visibility), do: query
 253
 254   defp restrict_since(query, %{"since_id" => since_id}) do
 255     from(activity in query, where: activity.id > ^since_id)
 256   end
 257
 258   defp restrict_since(query, _), do: query
 259
 260   defp restrict_tag(query, %{"tag" => tag}) do
 261     from(
 262       activity in query,
 263       where: fragment("? <@ (? #> '{\"object\",\"tag\"}')", ^tag, activity.data)
 264     )
 265   end
 266
 267   defp restrict_tag(query, _), do: query
 268
 269   defp restrict_recipients(query, [], _user), do: query
 270
 271   defp restrict_recipients(query, recipients, nil) do
 272     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 273   end
 274
 275   defp restrict_recipients(query, recipients, user) do
 276     from(
 277       activity in query,
 278       where: fragment("? && ?", ^recipients, activity.recipients),
 279       or_where: activity.actor == ^user.ap_id
 280     )
 281   end
 282
 283   defp restrict_limit(query, %{"limit" => limit}) do
 284     from(activity in query, limit: ^limit)
 285   end
 286
 287   defp restrict_limit(query, _), do: query
 288
 289   defp restrict_local(query, %{"local_only" => true}) do
 290     from(activity in query, where: activity.local == true)
 291   end
 292
 293   defp restrict_local(query, _), do: query
 294
 295   defp restrict_max(query, %{"max_id" => max_id}) do
 296     from(activity in query, where: activity.id < ^max_id)
 297   end
 298
 299   defp restrict_max(query, _), do: query
 300
 301   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 302     from(activity in query, where: activity.actor == ^actor_id)
 303   end
 304
 305   defp restrict_actor(query, _), do: query
 306
 307   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 308     restrict_type(query, %{"type" => [type]})
 309   end
 310
 311   defp restrict_type(query, %{"type" => type}) do
 312     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 313   end
 314
 315   defp restrict_type(query, _), do: query
 316
 317   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 318     from(
 319       activity in query,
 320       where: fragment("? <@ (? #> '{\"object\",\"likes\"}')", ^ap_id, activity.data)
 321     )
 322   end
 323
 324   defp restrict_favorited_by(query, _), do: query
 325
 326   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 327     from(
 328       activity in query,
 329       where: fragment("not (? #> '{\"object\",\"attachment\"}' = ?)", activity.data, ^[])
 330     )
 331   end
 332
 333   defp restrict_media(query, _), do: query
 334
 335   # Only search through last 100_000 activities by default
 336   defp restrict_recent(query, %{"whole_db" => true}), do: query
 337
 338   defp restrict_recent(query, _) do
 339     since = (Repo.aggregate(Activity, :max, :id) || 0) - 100_000
 340
 341     from(activity in query, where: activity.id > ^since)
 342   end
 343
 344   defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
 345     blocks = info["blocks"] || []
 346
 347     from(
 348       activity in query,
 349       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 350       where: fragment("not (?->'to' \\?| ?)", activity.data, ^blocks)
 351     )
 352   end
 353
 354   defp restrict_blocked(query, _), do: query
 355
 356   def fetch_activities_query(recipients, opts \\ %{}) do
 357     base_query =
 358       from(
 359         activity in Activity,
 360         limit: 20,
 361         order_by: [fragment("? desc nulls last", activity.id)]
 362       )
 363
 364     base_query
 365     |> restrict_recipients(recipients, opts["user"])
 366     |> restrict_tag(opts)
 367     |> restrict_since(opts)
 368     |> restrict_local(opts)
 369     |> restrict_limit(opts)
 370     |> restrict_max(opts)
 371     |> restrict_actor(opts)
 372     |> restrict_type(opts)
 373     |> restrict_favorited_by(opts)
 374     |> restrict_recent(opts)
 375     |> restrict_blocked(opts)
 376     |> restrict_media(opts)
 377     |> restrict_visibility(opts)
 378   end
 379
 380   def fetch_activities(recipients, opts \\ %{}) do
 381     fetch_activities_query(recipients, opts)
 382     |> Repo.all()
 383     |> Enum.reverse()
 384   end
 385
 386   def upload(file) do
 387     data = Upload.store(file)
 388     Repo.insert(%Object{data: data})
 389   end
 390
 391   def user_data_from_user_object(data) do
 392     avatar =
 393       data["icon"]["url"] &&
 394         %{
 395           "type" => "Image",
 396           "url" => [%{"href" => data["icon"]["url"]}]
 397         }
 398
 399     banner =
 400       data["image"]["url"] &&
 401         %{
 402           "type" => "Image",
 403           "url" => [%{"href" => data["image"]["url"]}]
 404         }
 405
 406     user_data = %{
 407       ap_id: data["id"],
 408       info: %{
 409         "ap_enabled" => true,
 410         "source_data" => data,
 411         "banner" => banner
 412       },
 413       avatar: avatar,
 414       nickname: "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}",
 415       name: data["name"],
 416       follower_address: data["followers"],
 417       bio: data["summary"]
 418     }
 419
 420     {:ok, user_data}
 421   end
 422
 423   def fetch_and_prepare_user_from_ap_id(ap_id) do
 424     with {:ok, %{status_code: 200, body: body}} <-
 425            @httpoison.get(ap_id, Accept: "application/activity+json"),
 426          {:ok, data} <- Jason.decode(body) do
 427       user_data_from_user_object(data)
 428     else
 429       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
 430     end
 431   end
 432
 433   def make_user_from_ap_id(ap_id) do
 434     if _user = User.get_by_ap_id(ap_id) do
 435       Transmogrifier.upgrade_user_from_ap_id(ap_id)
 436     else
 437       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
 438         User.insert_or_update_user(data)
 439       else
 440         e -> {:error, e}
 441       end
 442     end
 443   end
 444
 445   def make_user_from_nickname(nickname) do
 446     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
 447       make_user_from_ap_id(ap_id)
 448     else
 449       _e -> {:error, "No AP id in WebFinger"}
 450     end
 451   end
 452
 453   def publish(actor, activity) do
 454     followers =
 455       if actor.follower_address in activity.recipients do
 456         {:ok, followers} = User.get_followers(actor)
 457         followers |> Enum.filter(&(!&1.local))
 458       else
 459         []
 460       end
 461
 462     remote_inboxes =
 463       (Pleroma.Web.Salmon.remote_users(activity) ++ followers)
 464       |> Enum.filter(fn user -> User.ap_enabled?(user) end)
 465       |> Enum.map(fn %{info: %{"source_data" => data}} ->
 466         (data["endpoints"] && data["endpoints"]["sharedInbox"]) || data["inbox"]
 467       end)
 468       |> Enum.uniq()
 469
 470     {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
 471     json = Jason.encode!(data)
 472
 473     Enum.each(remote_inboxes, fn inbox ->
 474       Federator.enqueue(:publish_single_ap, %{
 475         inbox: inbox,
 476         json: json,
 477         actor: actor,
 478         id: activity.data["id"]
 479       })
 480     end)
 481   end
 482
 483   def publish_one(%{inbox: inbox, json: json, actor: actor, id: id}) do
 484     Logger.info("Federating #{id} to #{inbox}")
 485     host = URI.parse(inbox).host
 486
 487     signature =
 488       Pleroma.Web.HTTPSignatures.sign(actor, %{host: host, "content-length": byte_size(json)})
 489
 490     @httpoison.post(
 491       inbox,
 492       json,
 493       [{"Content-Type", "application/activity+json"}, {"signature", signature}],
 494       hackney: [pool: :default]
 495     )
 496   end
 497
 498   # TODO:
 499   # This will create a Create activity, which we need internally at the moment.
 500   def fetch_object_from_id(id) do
 501     if object = Object.get_cached_by_ap_id(id) do
 502       {:ok, object}
 503     else
 504       Logger.info("Fetching #{id} via AP")
 505
 506       with true <- String.starts_with?(id, "http"),
 507            {:ok, %{body: body, status_code: code}} when code in 200..299 <-
 508              @httpoison.get(
 509                id,
 510                [Accept: "application/activity+json"],
 511                follow_redirect: true,
 512                timeout: 10000,
 513                recv_timeout: 20000
 514              ),
 515            {:ok, data} <- Jason.decode(body),
 516            nil <- Object.get_by_ap_id(data["id"]),
 517            params <- %{
 518              "type" => "Create",
 519              "to" => data["to"],
 520              "cc" => data["cc"],
 521              "actor" => data["attributedTo"],
 522              "object" => data
 523            },
 524            {:ok, activity} <- Transmogrifier.handle_incoming(params) do
 525         {:ok, Object.get_by_ap_id(activity.data["object"]["id"])}
 526       else
 527         object = %Object{} ->
 528           {:ok, object}
 529
 530         _e ->
 531           Logger.info("Couldn't get object via AP, trying out OStatus fetching...")
 532
 533           case OStatus.fetch_activity_from_url(id) do
 534             {:ok, [activity | _]} -> {:ok, Object.get_by_ap_id(activity.data["object"]["id"])}
 535             e -> e
 536           end
 537       end
 538     end
 539   end
 540
 541   def is_public?(activity) do
 542     "https://www.w3.org/ns/activitystreams#Public" in (activity.data["to"] ++
 543                                                          (activity.data["cc"] || []))
 544   end
 545
 546   def visible_for_user?(activity, nil) do
 547     is_public?(activity)
 548   end
 549
 550   def visible_for_user?(activity, user) do
 551     x = [user.ap_id | user.following]
 552     y = activity.data["to"] ++ (activity.data["cc"] || [])
 553     visible_for_user?(activity, nil) || Enum.any?(x, &(&1 in y))
 554   end
 555 end