activitypub / includes / class-dispatcher.php

class-dispatcher.php in ActivityPub 8.0.2, at includes/class-dispatcher.php

557 lines 18.1 KB

Raw Download Zip

1	<?php
2	/**
3	* ActivityPub Dispatcher Class.
4	*
5	* @package Activitypub
6	*/
7
8	namespace Activitypub;
9
10	use Activitypub\Activity\Activity;
11	use Activitypub\Collection\Followers;
12	use Activitypub\Collection\Outbox;
13
14	/**
15	* ActivityPub Dispatcher Class.
16	*
17	* @author Matthias Pfefferle
18	*
19	* @see https://www.w3.org/TR/activitypub/
20	*/
21	class Dispatcher {
22	/**
23	* Batch size.
24	*
25	* @deprecated 7.6.0 Use {@see Dispatcher::get_batch_size()}.
26	*
27	* @var int
28	*/
29	public static $batch_size = ACTIVITYPUB_OUTBOX_PROCESSING_BATCH_SIZE;
30
31	/**
32	* Initialize the class, registering WordPress hooks.
33	*/
34	public static function init() {
35	\add_action( 'activitypub_process_outbox', array( self::class, 'process_outbox' ) );
36
37	\add_action( 'post_activitypub_add_to_outbox', array( self::class, 'send_immediate_accept' ), 10, 2 );
38
39	// Default filters to add Inboxes to sent to.
40	\add_filter( 'activitypub_additional_inboxes', array( self::class, 'add_inboxes_by_mentioned_actors' ), 10, 3 );
41	\add_filter( 'activitypub_additional_inboxes', array( self::class, 'add_inboxes_of_replied_urls' ), 10, 3 );
42	\add_filter( 'activitypub_additional_inboxes', array( self::class, 'add_inboxes_of_relays' ), 10, 3 );
43
44	Scheduler::register_async_batch_callback( 'activitypub_send_activity', array( self::class, 'send_to_followers' ) );
45	Scheduler::register_async_batch_callback( 'activitypub_retry_activity', array( self::class, 'retry_send_to_followers' ) );
46	}
47
48	/**
49	* Get the batch size for processing outbox items.
50	*
51	* @return int The batch size.
52	*/
53	public static function get_batch_size() {
54	/**
55	* Filters the batch size for processing outbox items.
56	*
57	* @param int $batch_size The batch size. Default ACTIVITYPUB_OUTBOX_PROCESSING_BATCH_SIZE.
58	*/
59	return apply_filters( 'activitypub_dispatcher_batch_size', ACTIVITYPUB_OUTBOX_PROCESSING_BATCH_SIZE );
60	}
61
62	/**
63	* Get the maximum number of retry attempts.
64	*
65	* @return int The maximum number of retry attempts.
66	*/
67	public static function get_retry_max_attempts() {
68	/**
69	* Filters the maximum number of retry attempts.
70	*
71	* @param int $retry_max_attempts The maximum number of retry attempts. Default ACTIVITYPUB_OUTBOX_RETRY_MAX_ATTEMPTS.
72	*/
73	return apply_filters( 'activitypub_dispatcher_retry_max_attempts', 3 );
74	}
75
76	/**
77	* Get the retry delay unit (in seconds).
78	*
79	* Used to calculate exponential backoff: time() + (attempt * attempt * retry_delay_unit).
80	*
81	* @return int The retry delay unit in seconds.
82	*/
83	public static function get_retry_delay() {
84	/**
85	* Filters the retry delay unit (in seconds).
86	*
87	* Used to calculate exponential backoff: time() + (attempt * attempt * retry_delay_unit).
88	*
89	* @param int $retry_delay_unit The retry delay unit in seconds. Default ACTIVITYPUB_OUTBOX_RETRY_DELAY_UNIT.
90	*/
91	return apply_filters( 'activitypub_dispatcher_retry_delay', HOUR_IN_SECONDS );
92	}
93
94	/**
95	* Get the error codes that qualify for a retry.
96	*
97	* @see https://github.com/tfredrich/RestApiTutorial.com/blob/fd08b0f67f07450521d143b123cd6e1846cb2e3b/content/advanced/responses/retries.md
98	*
99	* @return int[] The error codes.
100	*/
101	public static function get_retry_error_codes() {
102	/**
103	* Filters the error codes that qualify for a retry.
104	*
105	* @param int[] $retry_error_codes The error codes. Default array( 408, 429, 500, 502, 503, 504 ).
106	*/
107	return apply_filters( 'activitypub_dispatcher_retry_error_codes', ACTIVITYPUB_RETRY_ERROR_CODES );
108	}
109
110	/**
111	* Process the outbox.
112	*
113	* @param int $id The outbox ID.
114	*/
115	public static function process_outbox( $id ) {
116	$outbox_item = \get_post( $id );
117
118	// If the activity is not a post, return.
119	if ( ! $outbox_item ) {
120	return;
121	}
122
123	$type = \get_post_meta( $outbox_item->ID, '_activitypub_activity_type', true );
124	$actor = Outbox::get_actor( $outbox_item );
125	if ( \is_wp_error( $actor ) && 'Delete' !== $type ) {
126	// If the actor is not found, publish the post and don't try again.
127	\wp_publish_post( $outbox_item );
128	return;
129	}
130
131	$activity = Outbox::get_activity( $outbox_item );
132
133	// Send to mentioned and replied-to users. Everyone other than followers.
134	self::send_to_additional_inboxes( $activity, $outbox_item->post_author, $outbox_item );
135
136	if ( self::should_send_to_followers( $activity, $actor, $outbox_item ) ) {
137	\do_action(
138	'activitypub_send_activity',
139	$outbox_item->ID,
140	self::get_batch_size(),
141	\get_post_meta( $outbox_item->ID, '_activitypub_outbox_offset', true ) ?: 0 // phpcs:ignore
142	);
143	} else {
144	// No followers to process for this update. We're done.
145	\wp_publish_post( $outbox_item );
146	\delete_post_meta( $outbox_item->ID, '_activitypub_outbox_offset' );
147	}
148	}
149
150	/**
151	* Asynchronously runs batch processing routines.
152	*
153	* @param int $outbox_item_id The Outbox item ID.
154	* @param int\|null $batch_size Optional. The batch size. Default null (uses filtered batch size).
155	* @param int $offset Optional. The offset. Default 0.
156	*
157	* @return array\|void The next batch of followers to process, or void if done.
158	*/
159	public static function send_to_followers( $outbox_item_id, $batch_size = ACTIVITYPUB_OUTBOX_PROCESSING_BATCH_SIZE, $offset = 0 ) {
160	if ( null === $batch_size ) {
161	$batch_size = self::get_batch_size();
162	}
163
164	$outbox_item = \get_post( $outbox_item_id );
165	$json = Outbox::get_activity( $outbox_item_id )->to_json();
166	$inboxes = Followers::get_inboxes_for_activity( $json, $outbox_item->post_author, $batch_size, $offset );
167	$retries = self::send_to_inboxes( $inboxes, $outbox_item_id );
168
169	// Retry failed inboxes.
170	if ( ! empty( $retries ) ) {
171	self::schedule_retry( $retries, $outbox_item_id );
172	}
173
174	if ( is_countable( $inboxes ) && count( $inboxes ) < $batch_size ) {
175	\delete_post_meta( $outbox_item_id, '_activitypub_outbox_offset' );
176
177	/**
178	* Fires when the followers are complete.
179	*
180	* @param array $inboxes The inboxes.
181	* @param string $json The ActivityPub Activity JSON
182	* @param int $actor_id The actor ID.
183	* @param int $outbox_item_id The Outbox item ID.
184	* @param int $batch_size The batch size.
185	* @param int $offset The offset.
186	*/
187	\do_action( 'activitypub_outbox_processing_complete', $inboxes, $json, $outbox_item->post_author, $outbox_item_id, $batch_size, $offset );
188
189	// No more followers to process for this update.
190	\wp_publish_post( $outbox_item_id );
191	} else {
192	\update_post_meta( $outbox_item_id, '_activitypub_outbox_offset', $offset + $batch_size );
193
194	/**
195	* Fires when the batch of followers is complete.
196	*
197	* @param array $inboxes The inboxes.
198	* @param string $json The ActivityPub Activity JSON
199	* @param int $actor_id The actor ID.
200	* @param int $outbox_item_id The Outbox item ID.
201	* @param int $batch_size The batch size.
202	* @param int $offset The offset.
203	*/
204	\do_action( 'activitypub_outbox_processing_batch_complete', $inboxes, $json, $outbox_item->post_author, $outbox_item_id, $batch_size, $offset );
205
206	return array( $outbox_item_id, $batch_size, $offset + $batch_size );
207	}
208	}
209
210	/**
211	* Retry sending to followers.
212	*
213	* @param string $transient_key The key to retrieve retry inboxes.
214	* @param int $outbox_item_id The Outbox item ID.
215	* @param int $attempt The attempt number.
216	*/
217	public static function retry_send_to_followers( $transient_key, $outbox_item_id, $attempt = 1 ) {
218	$inboxes = \get_transient( $transient_key );
219	if ( false === $inboxes ) {
220	return;
221	}
222
223	// Delete the transient as we no longer need it.
224	\delete_transient( $transient_key );
225
226	$retries = self::send_to_inboxes( $inboxes, $outbox_item_id );
227
228	// Retry failed inboxes.
229	if ( ++$attempt < self::get_retry_max_attempts() && ! empty( $retries ) ) {
230	self::schedule_retry( $retries, $outbox_item_id, $attempt );
231	}
232	}
233
234	/**
235	* Send to inboxes.
236	*
237	* @param array $inboxes The inboxes to notify.
238	* @param int $outbox_item_id The Outbox item ID.
239	* @return array The failed inboxes.
240	*/
241	private static function send_to_inboxes( $inboxes, $outbox_item_id ) {
242	$outbox_item = \get_post( $outbox_item_id );
243
244	// Strip bto and bcc before delivery per ActivityPub spec Section 6.2.
245	\add_filter( 'activitypub_activity_object_array', array( self::class, 'strip_private_addressing' ) );
246	$json = Outbox::get_activity( $outbox_item_id )->to_json();
247	\remove_filter( 'activitypub_activity_object_array', array( self::class, 'strip_private_addressing' ) );
248
249	$retries = array();
250
251	/**
252	* Fires before sending an Activity to inboxes.
253	*
254	* @param string $json The ActivityPub Activity JSON.
255	* @param array $inboxes The inboxes to send to.
256	* @param int $outbox_item_id The Outbox item ID.
257	*/
258	\do_action( 'activitypub_pre_send_to_inboxes', $json, $inboxes, $outbox_item_id );
259
260	foreach ( $inboxes as $inbox ) {
261	// Handle local inboxes via internal REST API, remote via HTTP.
262	if ( is_same_domain( $inbox ) ) {
263	$result = self::send_to_local_inbox( $inbox, $json );
264	} else {
265	$result = safe_remote_post( $inbox, $json, $outbox_item->post_author );
266	}
267
268	if ( \is_wp_error( $result ) && in_array( $result->get_error_code(), self::get_retry_error_codes(), true ) ) {
269	$retries[] = $inbox;
270	}
271
272	/**
273	* Fires after an Activity has been sent to an inbox.
274	*
275	* @param array $result The result of the internal or remote post request.
276	* @param string $inbox The inbox URL.
277	* @param string $json The ActivityPub Activity JSON.
278	* @param int $actor_id The actor ID.
279	* @param int $outbox_item_id The Outbox item ID.
280	*/
281	\do_action( 'activitypub_sent_to_inbox', $result, $inbox, $json, $outbox_item->post_author, $outbox_item_id );
282	}
283
284	return $retries;
285	}
286
287	/**
288	* Send an activity to a local inbox via internal REST API request.
289	*
290	* @param string $inbox_url The local inbox URL.
291	* @param string $json The ActivityPub Activity JSON.
292	* @return array\|\WP_Error The result in the format of a remote post response, or WP_Error on failure.
293	*/
294	private static function send_to_local_inbox( $inbox_url, $json ) {
295	// Parse the inbox URL to extract the REST route.
296	$path = \wp_parse_url( $inbox_url, PHP_URL_PATH ) ?? '';
297	$rest_route = \preg_replace( '#^/' . preg_quote( \rest_get_url_prefix(), '#' ) . '#', '', $path );
298
299	// Create a REST request.
300	$request = new \WP_REST_Request( 'POST', $rest_route );
301	$request->set_header( 'Content-Type', 'application/activity+json' );
302	$request->set_body( $json );
303	$request->get_json_params();
304
305	\add_filter( 'activitypub_defer_signature_verification', '__return_true' );
306	$response = \rest_do_request( $request );
307	\remove_filter( 'activitypub_defer_signature_verification', '__return_true' );
308
309	// Return result in format similar to remote post response.
310	if ( $response->is_error() ) {
311	return $response->as_error();
312	}
313
314	return array(
315	'response' => array(
316	'code' => $response->get_status(),
317	),
318	'body' => \wp_json_encode( $response->get_data() ),
319	);
320	}
321
322	/**
323	* Strip bto and bcc fields from an Activity array before delivery.
324	*
325	* The ActivityPub spec (Section 6.2) requires servers to remove bto and bcc
326	* from Activities and their embedded objects before delivery to prevent
327	* revealing private recipient lists.
328	*
329	* Used as a temporary filter on `activitypub_activity_object_array` so that
330	* `to_json()` handles encoding consistently.
331	*
332	* @since 8.0.0
333	*
334	* @see https://www.w3.org/TR/activitypub/#delivery
335	*
336	* @param array $data The Activity array.
337	* @return array The sanitized array with bto and bcc removed.
338	*/
339	public static function strip_private_addressing( $data ) {
340	unset( $data['bto'], $data['bcc'] );
341
342	// Also strip from the embedded object, if present.
343	if ( isset( $data['object'] ) && \is_array( $data['object'] ) ) {
344	unset( $data['object']['bto'], $data['object']['bcc'] );
345	}
346
347	return $data;
348	}
349
350	/**
351	* Schedule a retry.
352	*
353	* @param array $retries The inboxes to retry.
354	* @param int $outbox_item_id The Outbox item ID.
355	* @param int $attempt Optional. The attempt number. Default 1.
356	*/
357	private static function schedule_retry( $retries, $outbox_item_id, $attempt = 1 ) {
358	$transient_key = 'activitypub_retry_' . \wp_generate_password( 12, false );
359	\set_transient( $transient_key, $retries, WEEK_IN_SECONDS );
360
361	\wp_schedule_single_event(
362	\time() + ( $attempt * $attempt * self::get_retry_delay() ),
363	'activitypub_retry_activity',
364	array( $transient_key, $outbox_item_id, $attempt )
365	);
366	}
367
368	/**
369	* Send an Activity to a custom list of inboxes, like mentioned users or replied-to posts.
370	*
371	* For all custom implementations, please use the `activitypub_additional_inboxes` filter.
372	*
373	* @param Activity $activity The ActivityPub Activity.
374	* @param int $actor_id The actor ID.
375	* @param \WP_Post $outbox_item The WordPress object.
376	*/
377	private static function send_to_additional_inboxes( $activity, $actor_id, $outbox_item = null ) {
378	/**
379	* Filters the list of inboxes to send the Activity to.
380	*
381	* @param array $inboxes The list of inboxes to send to.
382	* @param int $actor_id The actor ID.
383	* @param Activity $activity The ActivityPub Activity.
384	*/
385	$inboxes = apply_filters( 'activitypub_additional_inboxes', array(), $actor_id, $activity );
386	$inboxes = array_unique( $inboxes );
387
388	$retries = self::send_to_inboxes( $inboxes, $outbox_item->ID );
389
390	// Retry failed inboxes.
391	if ( ! empty( $retries ) ) {
392	self::schedule_retry( $retries, $outbox_item->ID );
393	}
394	}
395
396	/**
397	* Default filter to add Inboxes of Mentioned Actors
398	*
399	* @param array $inboxes The list of Inboxes.
400	* @param int $actor_id The WordPress Actor-ID.
401	* @param Activity $activity The ActivityPub Activity.
402	*
403	* @return array The filtered Inboxes.
404	*/
405	public static function add_inboxes_by_mentioned_actors( $inboxes, $actor_id, $activity ) {
406	$cc = $activity->get_cc() ?? array();
407	$to = $activity->get_to() ?? array();
408
409	$audience = array_merge( $cc, $to );
410
411	// Remove "public placeholder" from the audience.
412	$audience = array_diff( $audience, ACTIVITYPUB_PUBLIC_AUDIENCE_IDENTIFIERS );
413
414	if ( $audience ) {
415	$mentioned_inboxes = Mention::get_inboxes( $audience );
416
417	return array_merge( $inboxes, $mentioned_inboxes );
418	}
419
420	return $inboxes;
421	}
422
423	/**
424	* Default filter to add Inboxes of Posts that are set as `in-reply-to`
425	*
426	* @param array $inboxes The list of Inboxes.
427	* @param int $actor_id The WordPress Actor-ID.
428	* @param Activity $activity The ActivityPub Activity.
429	*
430	* @return array The filtered Inboxes
431	*/
432	public static function add_inboxes_of_replied_urls( $inboxes, $actor_id, $activity ) {
433	$in_reply_to = $activity->get_in_reply_to();
434
435	if ( ! $in_reply_to ) {
436	return $inboxes;
437	}
438
439	if ( ! is_array( $in_reply_to ) ) {
440	$in_reply_to = array( $in_reply_to );
441	}
442
443	foreach ( $in_reply_to as $url ) {
444	// No need to self-notify.
445	if ( is_same_domain( $url ) ) {
446	continue;
447	}
448
449	$object = Http::get_remote_object( $url );
450
451	if (
452	! $object \|\|
453	\is_wp_error( $object ) \|\|
454	empty( $object['attributedTo'] )
455	) {
456	continue;
457	}
458
459	$actor = object_to_uri( $object['attributedTo'] );
460	$actor = Http::get_remote_object( $actor );
461
462	if ( ! $actor \|\| \is_wp_error( $actor ) ) {
463	continue;
464	}
465
466	if ( ! empty( $actor['endpoints']['sharedInbox'] ) ) {
467	$inboxes[] = $actor['endpoints']['sharedInbox'];
468	} elseif ( ! empty( $actor['inbox'] ) ) {
469	$inboxes[] = $actor['inbox'];
470	}
471	}
472
473	return $inboxes;
474	}
475
476	/**
477	* Check if an Activity should be sent to followers.
478	*
479	* @param Activity $activity The Activity object.
480	* @param \Activitypub\Model\User\|\Activitypub\Model\Blog $actor The Actor object.
481	* @param \WP_Post $outbox_item The Outbox item.
482	*
483	* @return boolean True if the Activity should be sent to followers, false if not.
484	*/
485	protected static function should_send_to_followers( $activity, $actor, $outbox_item ) {
486	$cc = $activity->get_cc() ?? array();
487	$to = $activity->get_to() ?? array();
488
489	$audience = array_merge( $cc, $to );
490
491	$send = (
492	// Check if activity is public.
493	is_activity_public( $activity ) \|\|
494	// ...or check if follower endpoint is set.
495	in_array( $actor->get_followers(), $audience, true )
496	);
497
498	if ( $send ) {
499	$followers = Followers::get_inboxes_for_activity( $activity->to_json(), $outbox_item->post_author );
500
501	// Only send if there are followers to send to.
502	$send = ! is_countable( $followers ) \|\| 0 < count( $followers );
503	}
504
505	/**
506	* Filters whether to send an Activity to followers.
507	*
508	* @param bool $send_activity_to_followers Whether to send the Activity to followers.
509	* @param Activity $activity The ActivityPub Activity.
510	* @param int $actor_id The actor ID.
511	* @param \WP_Post $outbox_item The WordPress object.
512	*/
513	return apply_filters( 'activitypub_send_activity_to_followers', $send, $activity, $outbox_item->post_author, $outbox_item );
514	}
515
516	/**
517	* Add Inboxes of Relays.
518	*
519	* @param array $inboxes The list of Inboxes.
520	* @param int $actor_id The Actor-ID.
521	* @param Activity $activity The ActivityPub Activity.
522	*
523	* @return array The filtered Inboxes.
524	*/
525	public static function add_inboxes_of_relays( $inboxes, $actor_id, $activity ) {
526	// Check if activity is public.
527	if ( ! is_activity_public( $activity ) ) {
528	return $inboxes;
529	}
530
531	$relays = \get_option( 'activitypub_relays', array() );
532
533	if ( empty( $relays ) ) {
534	return $inboxes;
535	}
536
537	return array_merge( $inboxes, $relays );
538	}
539
540	/**
541	* Send an immediate Accept activity for the given Outbox item.
542	*
543	* @param int $outbox_id The Outbox item ID.
544	* @param Activity $activity The Activity that was just added to the Outbox.
545	*/
546	public static function send_immediate_accept( $outbox_id, $activity ) {
547	$outbox_item = \get_post( $outbox_id );
548
549	if ( ! $outbox_item \|\| 'Accept' !== $activity->get_type() ) {
550	return;
551	}
552
553	// Send to mentioned and replied-to users. Everyone other than followers.
554	self::send_to_additional_inboxes( $activity, $outbox_item->post_author, $outbox_item );
555	}
556	}
557