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 `; 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 _federation;
readonly IStringLocalizer _localizer;
readonly ILogger _logger;
public WebPushJob(IHttpClientFactory http, IVapid vapid, IServiceScopeFactory scopes, IInstanceSettings settings, ILocalActorService localActors,
IOptionsMonitor federation, IStringLocalizer localizer, ILogger 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 Handle(Job job, CancellationToken token)
{
var payload = JsonSerializer.Deserialize(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().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().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(), subscription, token))
{
await WebPush.ForgetToken(subscription.TokenId);
return JobOutcome.Done;
}
var avatar = await DB.Default.Find().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().Notifications(new List { notification }, avatarId, token))
.FirstOrDefault();
if (mapped == default)
return JobOutcome.Done;
json = JsonSerializer.SerializeToUtf8Bytes(new Dictionary
{
["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(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);
}
}
/// https, a host, no user information; plain http only on a test network (Federation:AllowPlainHttp, which
/// production refuses at start).
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 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(@"
|
\s*]*>", RegexOptions.IgnoreCase | RegexOptions.Compiled);
static readonly Regex Tags = new("<[^>]*>", RegexOptions.Compiled);
}
}