formatting
[akkoma] / 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.Object.Fetcher
4 alias Pleroma.Web.ActivityPub.{Transmogrifier, MRF}
5 alias Pleroma.Web.WebFinger
6 alias Pleroma.Web.Federator
7 alias Pleroma.Web.OStatus
8 import Ecto.Query
9 import Pleroma.Web.ActivityPub.Utils
10 require Logger
11
12 @httpoison Application.get_env(:pleroma, :httpoison)
13
14 # For Announce activities, we filter the recipients based on following status for any actors
15 # that match actual users. See issue #164 for more information about why this is necessary.
16 defp get_recipients(%{"type" => "Announce"} = data) do
17 to = data["to"] || []
18 cc = data["cc"] || []
19 recipients = to ++ cc
20 actor = User.get_cached_by_ap_id(data["actor"])
21
22 recipients
23 |> Enum.filter(fn recipient ->
24 case User.get_cached_by_ap_id(recipient) do
25 nil ->
26 true
27
28 user ->
29 User.following?(user, actor)
30 end
31 end)
32
33 {recipients, to, cc}
34 end
35
36 defp get_recipients(data) do
37 to = data["to"] || []
38 cc = data["cc"] || []
39 recipients = to ++ cc
40 {recipients, to, cc}
41 end
42
43 defp check_actor_is_active(actor) do
44 if not is_nil(actor) do
45 with user <- User.get_cached_by_ap_id(actor),
46 false <- user.info.deactivated do
47 :ok
48 else
49 _e -> :reject
50 end
51 else
52 :ok
53 end
54 end
55
56 def insert(map, local \\ true) when is_map(map) do
57 with nil <- Activity.normalize(map),
58 map <- lazy_put_activity_defaults(map),
59 :ok <- check_actor_is_active(map["actor"]),
60 {:ok, map} <- MRF.filter(map),
61 {:ok, map} <- insert_full_object(map) do
62 {recipients, _, _} = get_recipients(map)
63
64 {:ok, activity} =
65 Repo.insert(%Activity{
66 data: map,
67 local: local,
68 actor: map["actor"],
69 recipients: recipients
70 })
71
72 Notification.create_notifications(activity)
73 stream_out(activity)
74 {:ok, activity}
75 else
76 %Activity{} = activity -> {:ok, activity}
77 error -> {:error, error}
78 end
79 end
80
81 def stream_out(activity) do
82 public = "https://www.w3.org/ns/activitystreams#Public"
83
84 if activity.data["type"] in ["Create", "Announce"] do
85 object = Object.normalize(activity.data["object"])
86
87 Pleroma.Web.Streamer.stream("user", activity)
88 Pleroma.Web.Streamer.stream("list", activity)
89
90 if Enum.member?(activity.data["to"], public) do
91 Pleroma.Web.Streamer.stream("public", activity)
92
93 if activity.local do
94 Pleroma.Web.Streamer.stream("public:local", activity)
95 end
96
97 object.data
98 |> Map.get("tag", [])
99 |> Enum.filter(fn tag -> is_bitstring(tag) end)
100 |> Enum.map(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end)
101
102 if object.data["attachment"] != [] do
103 Pleroma.Web.Streamer.stream("public:media", activity)
104
105 if activity.local do
106 Pleroma.Web.Streamer.stream("public:local:media", activity)
107 end
108 end
109 else
110 if !Enum.member?(activity.data["cc"] || [], public) &&
111 !Enum.member?(
112 activity.data["to"],
113 User.get_by_ap_id(activity.data["actor"]).follower_address
114 ),
115 do: Pleroma.Web.Streamer.stream("direct", activity)
116 end
117 end
118 end
119
120 def create(%{to: to, actor: actor, context: context, object: object} = params) do
121 additional = params[:additional] || %{}
122 # only accept false as false value
123 local = !(params[:local] == false)
124 published = params[:published]
125
126 with create_data <-
127 make_create_data(
128 %{to: to, actor: actor, published: published, context: context, object: object},
129 additional
130 ),
131 {:ok, activity} <- insert(create_data, local),
132 :ok <- maybe_federate(activity),
133 {:ok, _actor} <- User.increase_note_count(actor) do
134 {:ok, activity}
135 end
136 end
137
138 def accept(%{to: to, actor: actor, object: object} = params) do
139 # only accept false as false value
140 local = !(params[:local] == false)
141
142 with data <- %{"to" => to, "type" => "Accept", "actor" => actor, "object" => object},
143 {:ok, activity} <- insert(data, local),
144 :ok <- maybe_federate(activity) do
145 {:ok, activity}
146 end
147 end
148
149 def reject(%{to: to, actor: actor, object: object} = params) do
150 # only accept false as false value
151 local = !(params[:local] == false)
152
153 with data <- %{"to" => to, "type" => "Reject", "actor" => actor, "object" => object},
154 {:ok, activity} <- insert(data, local),
155 :ok <- maybe_federate(activity) do
156 {:ok, activity}
157 end
158 end
159
160 def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
161 # only accept false as false value
162 local = !(params[:local] == false)
163
164 with data <- %{
165 "to" => to,
166 "cc" => cc,
167 "type" => "Update",
168 "actor" => actor,
169 "object" => object
170 },
171 {:ok, activity} <- insert(data, local),
172 :ok <- maybe_federate(activity) do
173 {:ok, activity}
174 end
175 end
176
177 # TODO: This is weird, maybe we shouldn't check here if we can make the activity.
178 def like(
179 %User{ap_id: ap_id} = user,
180 %Object{data: %{"id" => _}} = object,
181 activity_id \\ nil,
182 local \\ true
183 ) do
184 with nil <- get_existing_like(ap_id, object),
185 like_data <- make_like_data(user, object, activity_id),
186 {:ok, activity} <- insert(like_data, local),
187 {:ok, object} <- add_like_to_object(activity, object),
188 :ok <- maybe_federate(activity) do
189 {:ok, activity, object}
190 else
191 %Activity{} = activity -> {:ok, activity, object}
192 error -> {:error, error}
193 end
194 end
195
196 def unlike(
197 %User{} = actor,
198 %Object{} = object,
199 activity_id \\ nil,
200 local \\ true
201 ) do
202 with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
203 unlike_data <- make_unlike_data(actor, like_activity, activity_id),
204 {:ok, unlike_activity} <- insert(unlike_data, local),
205 {:ok, _activity} <- Repo.delete(like_activity),
206 {:ok, object} <- remove_like_from_object(like_activity, object),
207 :ok <- maybe_federate(unlike_activity) do
208 {:ok, unlike_activity, like_activity, object}
209 else
210 _e -> {:ok, object}
211 end
212 end
213
214 def announce(
215 %User{ap_id: _} = user,
216 %Object{data: %{"id" => _}} = object,
217 activity_id \\ nil,
218 local \\ true
219 ) do
220 with true <- is_public?(object),
221 announce_data <- make_announce_data(user, object, activity_id),
222 {:ok, activity} <- insert(announce_data, local),
223 {:ok, object} <- add_announce_to_object(activity, object),
224 :ok <- maybe_federate(activity) do
225 {:ok, activity, object}
226 else
227 error -> {:error, error}
228 end
229 end
230
231 def unannounce(
232 %User{} = actor,
233 %Object{} = object,
234 activity_id \\ nil,
235 local \\ true
236 ) do
237 with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
238 unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
239 {:ok, unannounce_activity} <- insert(unannounce_data, local),
240 :ok <- maybe_federate(unannounce_activity),
241 {:ok, _activity} <- Repo.delete(announce_activity),
242 {:ok, object} <- remove_announce_from_object(announce_activity, object) do
243 {:ok, unannounce_activity, object}
244 else
245 _e -> {:ok, object}
246 end
247 end
248
249 def follow(follower, followed, activity_id \\ nil, local \\ true) do
250 with data <- make_follow_data(follower, followed, activity_id),
251 {:ok, activity} <- insert(data, local),
252 :ok <- maybe_federate(activity) do
253 {:ok, activity}
254 end
255 end
256
257 def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
258 with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
259 {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
260 unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
261 {:ok, activity} <- insert(unfollow_data, local),
262 :ok <- maybe_federate(activity) do
263 {:ok, activity}
264 end
265 end
266
267 def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, local \\ true) do
268 user = User.get_cached_by_ap_id(actor)
269
270 data = %{
271 "type" => "Delete",
272 "actor" => actor,
273 "object" => id,
274 "to" => [user.follower_address, "https://www.w3.org/ns/activitystreams#Public"]
275 }
276
277 with {:ok, _} <- Object.delete(object),
278 {:ok, activity} <- insert(data, local),
279 :ok <- maybe_federate(activity),
280 {:ok, _actor} <- User.decrease_note_count(user) do
281 {:ok, activity}
282 end
283 end
284
285 def block(blocker, blocked, activity_id \\ nil, local \\ true) do
286 ap_config = Application.get_env(:pleroma, :activitypub)
287 unfollow_blocked = Keyword.get(ap_config, :unfollow_blocked)
288 outgoing_blocks = Keyword.get(ap_config, :outgoing_blocks)
289
290 with true <- unfollow_blocked do
291 follow_activity = fetch_latest_follow(blocker, blocked)
292
293 if follow_activity do
294 unfollow(blocker, blocked, nil, local)
295 end
296 end
297
298 with true <- outgoing_blocks,
299 block_data <- make_block_data(blocker, blocked, activity_id),
300 {:ok, activity} <- insert(block_data, local),
301 :ok <- maybe_federate(activity) do
302 {:ok, activity}
303 else
304 _e -> {:ok, nil}
305 end
306 end
307
308 def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
309 with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
310 unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
311 {:ok, activity} <- insert(unblock_data, local),
312 :ok <- maybe_federate(activity) do
313 {:ok, activity}
314 end
315 end
316
317 def fetch_activities_for_context(context, opts \\ %{}) do
318 public = ["https://www.w3.org/ns/activitystreams#Public"]
319
320 recipients =
321 if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
322
323 query = from(activity in Activity)
324
325 query =
326 query
327 |> restrict_blocked(opts)
328 |> restrict_recipients(recipients, opts["user"])
329
330 query =
331 from(
332 activity in query,
333 where:
334 fragment(
335 "?->>'type' = ? and ?->>'context' = ?",
336 activity.data,
337 "Create",
338 activity.data,
339 ^context
340 ),
341 order_by: [desc: :id]
342 )
343
344 Repo.all(query)
345 end
346
347 def fetch_public_activities(opts \\ %{}) do
348 q = fetch_activities_query(["https://www.w3.org/ns/activitystreams#Public"], opts)
349
350 q
351 |> restrict_unlisted()
352 |> Repo.all()
353 |> Enum.reverse()
354 end
355
356 @valid_visibilities ~w[direct unlisted public private]
357
358 defp restrict_visibility(query, %{visibility: "direct"}) do
359 public = "https://www.w3.org/ns/activitystreams#Public"
360
361 from(
362 activity in query,
363 join: sender in User,
364 on: sender.ap_id == activity.actor,
365 # Are non-direct statuses with no to/cc possible?
366 where:
367 fragment(
368 "not (? && ?)",
369 [^public, sender.follower_address],
370 activity.recipients
371 )
372 )
373 end
374
375 defp restrict_visibility(_query, %{visibility: visibility})
376 when visibility not in @valid_visibilities do
377 Logger.error("Could not restrict visibility to #{visibility}")
378 end
379
380 defp restrict_visibility(query, _visibility), do: query
381
382 def fetch_user_activities(user, reading_user, params \\ %{}) do
383 params =
384 params
385 |> Map.put("type", ["Create", "Announce"])
386 |> Map.put("actor_id", user.ap_id)
387 |> Map.put("whole_db", true)
388
389 recipients =
390 if reading_user do
391 ["https://www.w3.org/ns/activitystreams#Public"] ++
392 [reading_user.ap_id | reading_user.following]
393 else
394 ["https://www.w3.org/ns/activitystreams#Public"]
395 end
396
397 fetch_activities(recipients, params)
398 |> Enum.reverse()
399 end
400
401 defp restrict_since(query, %{"since_id" => since_id}) do
402 from(activity in query, where: activity.id > ^since_id)
403 end
404
405 defp restrict_since(query, _), do: query
406
407 defp restrict_tag(query, %{"tag" => tag}) do
408 from(
409 activity in query,
410 where: fragment("? <@ (? #> '{\"object\",\"tag\"}')", ^tag, activity.data)
411 )
412 end
413
414 defp restrict_tag(query, _), do: query
415
416 defp restrict_to_cc(query, recipients_to, recipients_cc) do
417 from(
418 activity in query,
419 where:
420 fragment(
421 "(?->'to' \\?| ?) or (?->'cc' \\?| ?)",
422 activity.data,
423 ^recipients_to,
424 activity.data,
425 ^recipients_cc
426 )
427 )
428 end
429
430 defp restrict_recipients(query, [], _user), do: query
431
432 defp restrict_recipients(query, recipients, nil) do
433 from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
434 end
435
436 defp restrict_recipients(query, recipients, user) do
437 from(
438 activity in query,
439 where: fragment("? && ?", ^recipients, activity.recipients),
440 or_where: activity.actor == ^user.ap_id
441 )
442 end
443
444 defp restrict_limit(query, %{"limit" => limit}) do
445 from(activity in query, limit: ^limit)
446 end
447
448 defp restrict_limit(query, _), do: query
449
450 defp restrict_local(query, %{"local_only" => true}) do
451 from(activity in query, where: activity.local == true)
452 end
453
454 defp restrict_local(query, _), do: query
455
456 defp restrict_max(query, %{"max_id" => max_id}) do
457 from(activity in query, where: activity.id < ^max_id)
458 end
459
460 defp restrict_max(query, _), do: query
461
462 defp restrict_actor(query, %{"actor_id" => actor_id}) do
463 from(activity in query, where: activity.actor == ^actor_id)
464 end
465
466 defp restrict_actor(query, _), do: query
467
468 defp restrict_type(query, %{"type" => type}) when is_binary(type) do
469 restrict_type(query, %{"type" => [type]})
470 end
471
472 defp restrict_type(query, %{"type" => type}) do
473 from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
474 end
475
476 defp restrict_type(query, _), do: query
477
478 defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
479 from(
480 activity in query,
481 where: fragment("? <@ (? #> '{\"object\",\"likes\"}')", ^ap_id, activity.data)
482 )
483 end
484
485 defp restrict_favorited_by(query, _), do: query
486
487 defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
488 from(
489 activity in query,
490 where: fragment("not (? #> '{\"object\",\"attachment\"}' = ?)", activity.data, ^[])
491 )
492 end
493
494 defp restrict_media(query, _), do: query
495
496 defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
497 from(
498 activity in query,
499 where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
500 )
501 end
502
503 defp restrict_replies(query, _), do: query
504
505 # Only search through last 100_000 activities by default
506 defp restrict_recent(query, %{"whole_db" => true}), do: query
507
508 defp restrict_recent(query, _) do
509 since = (Repo.aggregate(Activity, :max, :id) || 0) - 100_000
510
511 from(activity in query, where: activity.id > ^since)
512 end
513
514 defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
515 blocks = info.blocks || []
516 domain_blocks = info.domain_blocks || []
517
518 from(
519 activity in query,
520 where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
521 where: fragment("not (?->'to' \\?| ?)", activity.data, ^blocks),
522 where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks)
523 )
524 end
525
526 defp restrict_blocked(query, _), do: query
527
528 defp restrict_unlisted(query) do
529 from(
530 activity in query,
531 where:
532 fragment(
533 "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
534 activity.data,
535 ^["https://www.w3.org/ns/activitystreams#Public"]
536 )
537 )
538 end
539
540 def fetch_activities_query(recipients, opts \\ %{}) do
541 base_query =
542 from(
543 activity in Activity,
544 limit: 20,
545 order_by: [fragment("? desc nulls last", activity.id)]
546 )
547
548 base_query
549 |> restrict_recipients(recipients, opts["user"])
550 |> restrict_tag(opts)
551 |> restrict_since(opts)
552 |> restrict_local(opts)
553 |> restrict_limit(opts)
554 |> restrict_max(opts)
555 |> restrict_actor(opts)
556 |> restrict_type(opts)
557 |> restrict_favorited_by(opts)
558 |> restrict_recent(opts)
559 |> restrict_blocked(opts)
560 |> restrict_media(opts)
561 |> restrict_visibility(opts)
562 |> restrict_replies(opts)
563 end
564
565 def fetch_activities(recipients, opts \\ %{}) do
566 fetch_activities_query(recipients, opts)
567 |> Repo.all()
568 |> Enum.reverse()
569 end
570
571 def fetch_activities_bounded(recipients_to, recipients_cc, opts \\ %{}) do
572 fetch_activities_query([], opts)
573 |> restrict_to_cc(recipients_to, recipients_cc)
574 |> Repo.all()
575 |> Enum.reverse()
576 end
577
578 def upload(file, opts \\ []) do
579 with {:ok, data} <- Upload.store(file, opts) do
580 Repo.insert(%Object{data: data})
581 end
582 end
583
584 def user_data_from_user_object(data) do
585 avatar =
586 data["icon"]["url"] &&
587 %{
588 "type" => "Image",
589 "url" => [%{"href" => data["icon"]["url"]}]
590 }
591
592 banner =
593 data["image"]["url"] &&
594 %{
595 "type" => "Image",
596 "url" => [%{"href" => data["image"]["url"]}]
597 }
598
599 locked = data["manuallyApprovesFollowers"] || false
600 data = Transmogrifier.maybe_fix_user_object(data)
601
602 user_data = %{
603 ap_id: data["id"],
604 info: %{
605 "ap_enabled" => true,
606 "source_data" => data,
607 "banner" => banner,
608 "locked" => locked
609 },
610 avatar: avatar,
611 name: data["name"],
612 follower_address: data["followers"],
613 bio: data["summary"]
614 }
615
616 # nickname can be nil because of virtual actors
617 user_data =
618 if data["preferredUsername"] do
619 Map.put(
620 user_data,
621 :nickname,
622 "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
623 )
624 else
625 Map.put(user_data, :nickname, nil)
626 end
627
628 {:ok, user_data}
629 end
630
631 def fetch_and_prepare_user_from_ap_id(ap_id) do
632 with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id) do
633 user_data_from_user_object(data)
634 else
635 e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
636 end
637 end
638
639 def make_user_from_ap_id(ap_id) do
640 if _user = User.get_by_ap_id(ap_id) do
641 Transmogrifier.upgrade_user_from_ap_id(ap_id)
642 else
643 with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
644 User.insert_or_update_user(data)
645 else
646 e -> {:error, e}
647 end
648 end
649 end
650
651 def make_user_from_nickname(nickname) do
652 with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
653 make_user_from_ap_id(ap_id)
654 else
655 _e -> {:error, "No AP id in WebFinger"}
656 end
657 end
658
659 def should_federate?(inbox, public) do
660 if public do
661 true
662 else
663 inbox_info = URI.parse(inbox)
664 !Enum.member?(Pleroma.Config.get([:instance, :quarantined_instances], []), inbox_info.host)
665 end
666 end
667
668 def publish(actor, activity) do
669 followers =
670 if actor.follower_address in activity.recipients do
671 {:ok, followers} = User.get_followers(actor)
672 followers |> Enum.filter(&(!&1.local))
673 else
674 []
675 end
676
677 public = is_public?(activity)
678
679 remote_inboxes =
680 (Pleroma.Web.Salmon.remote_users(activity) ++ followers)
681 |> Enum.filter(fn user -> User.ap_enabled?(user) end)
682 |> Enum.map(fn %{info: %{source_data: data}} ->
683 (is_map(data["endpoints"]) && Map.get(data["endpoints"], "sharedInbox")) || data["inbox"]
684 end)
685 |> Enum.uniq()
686 |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
687
688 {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
689 json = Jason.encode!(data)
690
691 Enum.each(remote_inboxes, fn inbox ->
692 Federator.enqueue(:publish_single_ap, %{
693 inbox: inbox,
694 json: json,
695 actor: actor,
696 id: activity.data["id"]
697 })
698 end)
699 end
700
701 def publish_one(%{inbox: inbox, json: json, actor: actor, id: id}) do
702 Logger.info("Federating #{id} to #{inbox}")
703 host = URI.parse(inbox).host
704
705 digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64())
706
707 signature =
708 Pleroma.Web.HTTPSignatures.sign(actor, %{
709 host: host,
710 "content-length": byte_size(json),
711 digest: digest
712 })
713
714 @httpoison.post(
715 inbox,
716 json,
717 [
718 {"Content-Type", "application/activity+json"},
719 {"signature", signature},
720 {"digest", digest}
721 ],
722 hackney: [pool: :default]
723 )
724 end
725
726 def is_public?(activity) do
727 "https://www.w3.org/ns/activitystreams#Public" in (activity.data["to"] ++
728 (activity.data["cc"] || []))
729 end
730
731 def visible_for_user?(activity, nil) do
732 is_public?(activity)
733 end
734
735 def visible_for_user?(activity, user) do
736 x = [user.ap_id | user.following]
737 y = activity.data["to"] ++ (activity.data["cc"] || [])
738 visible_for_user?(activity, nil) || Enum.any?(x, &(&1 in y))
739 end
740
741 # guard
742 def entire_thread_visible_for_user?(nil, user), do: false
743
744 # child / root
745 def entire_thread_visible_for_user?(
746 %Activity{data: %{"object" => object_id}} = tail,
747 user
748 ) do
749 parent = Activity.get_in_reply_to_activity(tail)
750
751 cond do
752 !is_nil(parent) ->
753 visible_for_user?(tail, user) && entire_thread_visible_for_user?(parent, user)
754
755 true ->
756 visible_for_user?(tail, user)
757 end
758 end
759
760 # filter out broken threads
761 def contain_broken_threads(%Activity{} = activity, %User{} = user) do
762 entire_thread_visible_for_user?(activity, user)
763 end
764
765 # do post-processing on a specific activity
766 def contain_activity(%Activity{} = activity, %User{} = user) do
767 contain_broken_threads(activity, user)
768 end
769
770 # do post-processing on a timeline
771 def contain_timeline(timeline, user) do
772 timeline
773 |> Enum.filter(fn activity ->
774 contain_activity(activity, user)
775 end)
776 end
777 end