SocialPub/PrivaPub/Infrastructure/Push/WebPushJob.cs

using Microsoft.Extensions.Localization;
using Microsoft.Extensions.Options;

using MongoDB.Entities;

using OpenIddict.Abstractions;

using PrivaPub.Api.Mastodon.Controllers;
using PrivaPub.Api.Mastodon.Mappers;
using PrivaPub.Domain.Instance;
using PrivaPub.Domain.Moderation;
using PrivaPub.Domain.Social;
using PrivaPub.Federation.Actors;
using PrivaPub.Infrastructure.Http;
using PrivaPub.Infrastructure.Jobs;
using PrivaPub.Models.Jobs;
using PrivaPub.Models.Social;
using PrivaPub.Models.User;
using PrivaPub.Resources;

using System.Net;
using System.Net.Http.Headers;
using System.Net.Sockets;
using System.Text.Json;
using System.Text.RegularExpressions;

using static OpenIddict.Abstractions.OpenIddictConstants;

namespace PrivaPub.Infrastructure.Push
{
	// Sends one Web Push notification to one subscription (JobKind.SendPush, queued by WebPushDispatcher; completion
	// program R9), as Mastodon's Web::PushNotificationWorker does:
	// - the payload is Mastodon's (access_token, preferred_locale, notification_id, notification_type, icon, title, body), built
	//   when it is sent, as the persona sees the notification then (one it can no longer see, or dismissed meanwhile, is not
	//   sent), its access_token the subscription's own: the token is checked to be valid and the persona's first;
	// - encrypted for the client alone (WebPushCrypto): RFC 8291 aes128gcm with RFC 8292's `vapid t=, k=` for a standard
	//   subscription, otherwise the aesgcm draft with Encryption and Crypto-Key and `WebPush <jwt>`; the JWT (Vapid) names the
	//   push service's origin (aud), lasts twelve hours (exp) and the server's contact (sub: mailto: the contact email, or
	//   the server's address; never a persona);
	// - TTL two days and Urgency normal, as Mastodon's;
	// - through the `push` client: public addresses only, https only, no redirects, small timeouts (the federation client's
	//   safe handler, with the pasture's exceptions where Federation says so).
	// A 404 or 410 means the subscription is gone: it is deleted. 408, 429, a 5xx, a timeout or a broken connection retry on
	// the jobs' backoff; any other answer drops the push (logged by the push service's host alone).
	public class WebPushJob : IJobHandler
	{
		public const string ClientName = "push";
		public static readonly TimeSpan RequestTimeout = TimeSpan.FromSeconds(15);
		public static readonly TimeSpan ConnectTimeout = TimeSpan.FromSeconds(5);
		// Mastodon's Web::PushNotificationWorker::TTL and URGENCY
		const int TimeToLive = 48 * 3600;
		const string Urgency = "normal";
		const int BodyLength = 140;

		readonly IHttpClientFactory _http;
		readonly IVapid _vapid;
		readonly IServiceScopeFactory _scopes;
		readonly IInstanceSettings _settings;
		readonly ILocalActorService _localActors;
		readonly IOptionsMonitor<FederationOptions> _federation;
		readonly IStringLocalizer<GenericRes> _localizer;
		readonly ILogger<WebPushJob> _logger;

		public WebPushJob(IHttpClientFactory http, IVapid vapid, IServiceScopeFactory scopes, IInstanceSettings settings, ILocalActorService localActors,
			IOptionsMonitor<FederationOptions> federation, IStringLocalizer<GenericRes> localizer, ILogger<WebPushJob> logger)
		{
			_http = http;
			_vapid = vapid;
			_scopes = scopes;
			_settings = settings;
			_localActors = localActors;
			_federation = federation;
			_localizer = localizer;
			_logger = logger;
		}

		public JobKind Kind => JobKind.SendPush;
		public int Concurrency => 4;
		public int MaxAttempts => 6;
		public int PerHostLimit => 4;

		public async Task<JobOutcome> Handle(Job job, CancellationToken token)
		{
			var payload = JsonSerializer.Deserialize<PushPayload>(job.Payload ?? "null");
			if (string.IsNullOrEmpty(payload?.SubscriptionId) || string.IsNullOrEmpty(payload.NotificationId))
				return JobOutcome.Done;
			var subscriptionId = payload.SubscriptionId;
			var subscription = await DB.Default.Find<PushSubscription>().Match(s => s.ID == subscriptionId).ExecuteFirstAsync(token);
			if (subscription == default)
				return JobOutcome.Done;
			// a persona held back (suspended, its login banned) gets no push
			if (Restrictions.IsSuspended(subscription.AvatarId))
				return JobOutcome.Done;
			var notificationId = payload.NotificationId;
			var avatarId = subscription.AvatarId;
			var notification = await DB.Default.Find<Notification>().Match(n => n.ID == notificationId && n.AvatarId == avatarId).ExecuteFirstAsync(token);
			if (notification == default || !await WebPush.Pushable(subscription, notification, token))
				return JobOutcome.Done;
			if (!Uri.TryCreate(subscription.Endpoint, UriKind.Absolute, out var endpoint) || !IsAllowed(endpoint, _federation.CurrentValue))
				return JobOutcome.Dead("endpoint");

			byte[] json;
			using (var scope = _scopes.CreateScope())
			{
				// the token the subscription was made with must still be valid, and the persona's: it goes into the payload
				if (!await TokenHolds(scope.ServiceProvider.GetRequiredService<IOpenIddictTokenManager>(), subscription, token))
				{
					await WebPush.ForgetToken(subscription.TokenId);
					return JobOutcome.Done;
				}
				var avatar = await DB.Default.Find<Avatar>().Match(a => a.ID == avatarId && a.DeletionAt == null).ExecuteFirstAsync(token);
				if (avatar == default)
				{
					await WebPush.Forget(new[] { avatarId });
					return JobOutcome.Done;
				}
				var mapped = (await scope.ServiceProvider.GetRequiredService<MastodonMapper>().Notifications(new List<Notification> { notification }, avatarId, token))
					.FirstOrDefault();
				if (mapped == default)
					return JobOutcome.Done;
				json = JsonSerializer.SerializeToUtf8Bytes(new Dictionary<string, object>
				{
					["access_token"] = subscription.AccessToken,
					["preferred_locale"] = Locale(avatar, scope.ServiceProvider),
					["notification_id"] = notification.ID,
					["notification_type"] = WebPush.AlertOf(notification.Type),
					["icon"] = mapped.Account?.AvatarStatic ?? mapped.Account?.Avatar,
					["title"] = Title(notification, mapped),
					["body"] = Body(mapped)
				});
			}

			var key = await _vapid.Key(token);
			var p256dh = WebPushCrypto.FromBase64Url(subscription.P256dh);
			var auth = WebPushCrypto.FromBase64Url(subscription.Auth);
			if (!WebPushCrypto.AreUsable(p256dh, auth))
				return JobOutcome.Dead("keys");
			var message = subscription.Standard ? WebPushCrypto.Aes128Gcm(json, p256dh, auth) : WebPushCrypto.AesGcm(json, p256dh, auth);
			var jwt = key.Token(VapidKey.Audience(endpoint), Subject(), DateTime.UtcNow);

			using var request = new HttpRequestMessage(HttpMethod.Post, endpoint) { Content = new ByteArrayContent(message.Body) };
			request.Content.Headers.ContentType = new MediaTypeHeaderValue("application/octet-stream");
			request.Content.Headers.ContentEncoding.Add(message.ContentEncoding);
			request.Headers.TryAddWithoutValidation("TTL", TimeToLive.ToString(System.Globalization.CultureInfo.InvariantCulture));
			request.Headers.TryAddWithoutValidation("Urgency", Urgency);
			if (subscription.Standard)
				request.Headers.TryAddWithoutValidation("Authorization", $"vapid t={jwt}, k={key.ForHeader}");
			else
			{
				request.Headers.TryAddWithoutValidation("Authorization", $"WebPush {jwt}");
				request.Headers.TryAddWithoutValidation("Encryption", $"salt={WebPushCrypto.Base64Url(message.Salt)}");
				request.Headers.TryAddWithoutValidation("Crypto-Key", $"dh={WebPushCrypto.Base64Url(message.ServerPublicKey)};p256ecdsa={key.ForHeader}");
			}

			try
			{
				using var response = await _http.CreateClient(ClientName).SendAsync(request, HttpCompletionOption.ResponseHeadersRead, token);
				var status = (int)response.StatusCode;
				if (status is >= 200 and < 300)
					return JobOutcome.Done;
				if (response.StatusCode is HttpStatusCode.NotFound or HttpStatusCode.Gone)
				{
					await DB.Default.DeleteAsync<PushSubscription>(subscription.ID);
					_logger.LogInformation("{Kind}: {Host} answered {Status}, the subscription is gone", Kind, endpoint.Host, status);
					return JobOutcome.Done;
				}
				if (response.StatusCode is HttpStatusCode.RequestTimeout or HttpStatusCode.TooManyRequests || status >= 500)
					return JobOutcome.Retry($"status {status}");
				_logger.LogWarning("{Kind}: {Host} refused a push with {Status}; it is dropped", Kind, endpoint.Host, status);
				return JobOutcome.Dead($"status {status}");
			}
			catch (BlockedDestinationException)
			{
				_logger.LogWarning("{Kind}: {Host} does not resolve to a public address; the push is dropped", Kind, endpoint.Host);
				return JobOutcome.Dead("blocked destination");
			}
			catch (TaskCanceledException) when (!token.IsCancellationRequested)
			{
				return JobOutcome.Retry("timeout");
			}
			catch (HttpRequestException ex) when (ex.InnerException is BlockedDestinationException)
			{
				_logger.LogWarning("{Kind}: {Host} does not resolve to a public address; the push is dropped", Kind, endpoint.Host);
				return JobOutcome.Dead("blocked destination");
			}
			catch (Exception ex) when (ex is HttpRequestException or IOException or SocketException)
			{
				return JobOutcome.Retry(ex.GetType().Name);
			}
		}

		/// <summary>https, a host, no user information; plain http only on a test network (Federation:AllowPlainHttp, which
		/// production refuses at start).</summary>
		public static bool IsAllowed(Uri endpoint, FederationOptions options) =>
			endpoint != default && endpoint.IsAbsoluteUri && !string.IsNullOrEmpty(endpoint.Host) && string.IsNullOrEmpty(endpoint.UserInfo)
			&& (endpoint.Scheme == Uri.UriSchemeHttps || options?.AllowPlainHttp == true && endpoint.Scheme == Uri.UriSchemeHttp);

		static async Task<bool> TokenHolds(IOpenIddictTokenManager tokens, PushSubscription subscription, CancellationToken token)
		{
			if (string.IsNullOrEmpty(subscription.TokenId) || string.IsNullOrEmpty(subscription.AccessToken))
				return false;
			var stored = await tokens.FindByIdAsync(subscription.TokenId, token);
			if (stored == default || !await tokens.HasStatusAsync(stored, Statuses.Valid, token))
				return false;
			if (await tokens.GetExpirationDateAsync(stored, token) is { } expires && expires < DateTimeOffset.UtcNow)
				return false;
			return await tokens.GetSubjectAsync(stored, token) == subscription.AvatarId;
		}

		// RFC 8292's sub: how the push service reaches the server's operator. The contact email the administrator gave, or
		// the server's own address; never a persona. (A server on plain http, a test network, has no https: address to give,
		// so it names a mailbox at its host: RFC 8292 takes only mailto: and https:.)
		string Subject()
		{
			var email = _settings.Current.ContactEmail?.Trim();
			if (!string.IsNullOrEmpty(email) && !email.Any(char.IsWhiteSpace) && email.Contains('@'))
				return "mailto:" + email;
			var address = _localActors.BaseAddress;
			return Uri.TryCreate(address, UriKind.Absolute, out var base_) && base_.Scheme != Uri.UriSchemeHttps
				? "mailto:privapub@" + base_.Host
				: address;
		}

		// the persona's language (its default for posts), else the server's first
		static string Locale(Avatar avatar, IServiceProvider services)
		{
			var language = avatar.Settings?.DefaultLanguage;
			return string.IsNullOrWhiteSpace(language) ? InstanceController.LanguageCodes(services)[0] : language.Trim();
		}

		// Mastodon's notification_mailer subjects
		string Title(Notification notification, Api.Mastodon.Entities.Notification mapped)
		{
			var name = string.IsNullOrWhiteSpace(mapped.Account?.DisplayName) ? mapped.Account?.Username ?? string.Empty : mapped.Account.DisplayName;
			return notification.Type switch
			{
				NotificationType.Mention => string.Format(_localizer["You were mentioned by {0}"], name),
				NotificationType.Status => string.Format(_localizer["{0} just posted"], name),
				NotificationType.Reblog => string.Format(_localizer["{0} boosted your post"], name),
				NotificationType.Follow => string.Format(_localizer["{0} is now following you"], name),
				NotificationType.FollowRequest => string.Format(_localizer["Pending follower: {0}"], name),
				NotificationType.Favourite => string.Format(_localizer["{0} favorited your post"], name),
				NotificationType.Poll => string.Format(_localizer["A poll by {0} has ended"], name),
				NotificationType.Update => string.Format(_localizer["{0} edited a post"], name),
				NotificationType.Quote => string.Format(_localizer["{0} quoted your post"], name),
				NotificationType.Reaction => string.Format(_localizer["{0} reacted to your post with {1}"], name, notification.Emoji),
				NotificationType.ModerationWarning => _localizer["You have received a moderation warning"],
				_ => name
			};
		}

		// Mastodon's: the post's warning, else its words, else the account's bio, without tags, 140 characters at most
		static string Body(Api.Mastodon.Entities.Notification mapped)
		{
			var source = !string.IsNullOrWhiteSpace(mapped.Status?.SpoilerText) ? mapped.Status.SpoilerText
				: mapped.Status?.Content ?? mapped.ModerationWarning?.Text ?? mapped.Account?.Note ?? string.Empty;
			var text = WebUtility.HtmlDecode(Tags.Replace(Breaks.Replace(source, "\n"), string.Empty)).Trim();
			return text.Length <= BodyLength ? text : text[..(BodyLength - 3)] + "...";
		}

		static readonly Regex Breaks = new(@"<br\s*/?>|</p>\s*<p[^>]*>", RegexOptions.IgnoreCase | RegexOptions.Compiled);
		static readonly Regex Tags = new("<[^>]*>", RegexOptions.Compiled);
	}
}