using PrivaPub.Models.Federation; using System.Collections.Concurrent; using Microsoft.Extensions.Options; using MongoDB.Entities; using PrivaPub.Federation.Actors; using PrivaPub.Infrastructure.Http; using PrivaPub.Models.Media; using System.Security.Cryptography; using System.Text; namespace PrivaPub.Domain.Media { public enum ProxyOutcome { Cached, TooBig, Failed } public interface IMediaProxy { string Wrap(string remoteUrl); string Verified(string signature, string encodedUrl); /// Whether a remote file's server is suspended or has its media rejected: nothing of it is proxied. bool Refuses(string url); (string Path, string ContentType) Cached(string url); /// Downloads a remote file into the cache, once however many ask at the same time. Task<(ProxyOutcome Outcome, string Path, string ContentType)> Download(string url, CancellationToken token); Task<(string Path, string ContentType)> Fetch(string signature, string encodedUrl, CancellationToken token); Task Open(string url, System.Net.Http.Headers.RangeHeaderValue range, CancellationToken token); /// Deletes what the cache holds of a domain and its subdomains; how many files. int Purge(string domain); /// Trims the cache to its size, oldest first. void Trim(); } // Remote media, fetched for clients so that they never contact a remote server (CLAUDE.md, media invariants): // - URLs are HMAC-signed, so only what PrivaPub showed is fetched; // - a download is shared by everyone asking for the same URL at once, streamed into a temporary file beside its place // and renamed there (a reader never sees half a file), and never held in memory; // - what is too big to cache is remembered for an hour (it is streamed instead), what failed for five minutes; // - the cache's size is counted as it grows, and trimmed as soon as it passes the cap; // - nothing of a suspended server, or of one whose media are rejected, is proxied, and blocking one purges its files. public class MediaProxy : IMediaProxy { static readonly TimeSpan TooBigFor = TimeSpan.FromHours(1); static readonly TimeSpan FailedFor = TimeSpan.FromMinutes(5); const int Downloads = 8;//remote files fetched at once, however many clients ask readonly ILocalActorService _localActors; readonly IFederationHttp _http; readonly IMediaService _media; readonly IOptionsMonitor _options; readonly Federation.Moderation.IDomainBlocks _domainBlocks; // Lazy: GetOrAdd may run its factory more than once when asked at the same time, but only one Lazy is ever evaluated readonly ConcurrentDictionary>> _inFlight = new(); readonly ConcurrentDictionary _refused = new(); readonly SemaphoreSlim _downloads = new(Downloads); readonly object _keyLock = new(); byte[] _key; long _bytes = -1;//what the cache holds, counted once from the disk and then as it changes int _trimming; public MediaProxy(ILocalActorService localActors, IFederationHttp http, IMediaService media, IOptionsMonitor options, Federation.Moderation.IDomainBlocks domainBlocks = default) { _localActors = localActors; _http = http; _media = media; _options = options; _domainBlocks = domainBlocks; } byte[] Key { get { if (_key != default) return _key; lock (_keyLock) return _key ??= LoadKey(); } } public string Wrap(string remoteUrl) { if (string.IsNullOrEmpty(remoteUrl) || remoteUrl.StartsWith(_localActors.BaseAddress + "/", StringComparison.OrdinalIgnoreCase)) return remoteUrl; var encoded = Base64Url(Encoding.UTF8.GetBytes(remoteUrl)); return $"{_localActors.BaseAddress}/media/proxy/{Sign(remoteUrl)}/{encoded}"; } public string Verified(string signature, string encodedUrl) { string url; try { url = Encoding.UTF8.GetString(FromBase64Url(encodedUrl)); } catch (FormatException) { return default; } return CryptographicOperations.FixedTimeEquals(Encoding.ASCII.GetBytes(signature ?? string.Empty), Encoding.ASCII.GetBytes(Sign(url))) ? url : default; } public bool Refuses(string url) { if (_domainBlocks == default || !Uri.TryCreate(url, UriKind.Absolute, out var target)) return false; var block = _domainBlocks.Find(target.Host); return block is { Severity: DomainBlockSeverity.Suspend } or { RejectMedia: true }; } public (string Path, string ContentType) Cached(string url) { var (path, typePath) = CachePaths(url); try { if (!File.Exists(path) || !File.Exists(typePath)) return default; File.SetLastWriteTimeUtc(path, DateTime.UtcNow); return (path, File.ReadLines(typePath).FirstOrDefault()); } catch (IOException) { return default;//trimmed meanwhile } } public Task Open(string url, System.Net.Http.Headers.RangeHeaderValue range, CancellationToken token) => _http.OpenMedia(url, range, token); public async Task<(string Path, string ContentType)> Fetch(string signature, string encodedUrl, CancellationToken token) { var url = Verified(signature, encodedUrl); if (url == default || Refuses(url)) return default; var (outcome, path, contentType) = await Download(url, token); return outcome == ProxyOutcome.Cached ? (path, contentType) : default; } public async Task<(ProxyOutcome Outcome, string Path, string ContentType)> Download(string url, CancellationToken token) { if (Cached(url) is { Path: not null } cached) return (ProxyOutcome.Cached, cached.Path, cached.ContentType); if (_refused.TryGetValue(url, out var refused)) { if (refused.Until > DateTime.UtcNow) return (refused.Outcome, default, default); _refused.TryRemove(url, out _); } // shared by everyone asking for it now; the download itself isn't cancelled when one of them leaves var shared = _inFlight.GetOrAdd(url, key => new Lazy>(() => DownloadOnce(key))); try { return await shared.Value.WaitAsync(token); } finally { if (shared.Value.IsCompleted) _inFlight.TryRemove(new KeyValuePair>>(url, shared)); } } async Task<(ProxyOutcome, string, string)> DownloadOnce(string url) { await Task.Yield(); if (!await _downloads.WaitAsync(TimeSpan.FromSeconds(30))) return (ProxyOutcome.Failed, default, default);//too busy: the client tries again // cached by a download that ended just before this one began if (Cached(url) is { Path: not null } done) { _downloads.Release(); return (ProxyOutcome.Cached, done.Path, done.ContentType); } var (path, typePath) = CachePaths(url); var part = $"{path}.{Guid.NewGuid():N}.part"; try { Directory.CreateDirectory(System.IO.Path.GetDirectoryName(path)!); string contentType, refusal; await using (var file = new FileStream(part, FileMode.CreateNew, FileAccess.Write, FileShare.None, 64 * 1024, useAsync: true)) (contentType, refusal) = await _http.DownloadMedia(url, _options.CurrentValue.MaxProxiedBytes, file, CancellationToken.None); if (contentType == default) { var outcome = refusal == "too-large" ? ProxyOutcome.TooBig : ProxyOutcome.Failed; _refused[url] = (outcome, DateTime.UtcNow + (outcome == ProxyOutcome.TooBig ? TooBigFor : FailedFor)); return (outcome, default, default); } await File.WriteAllTextAsync(typePath, contentType + "\n" + new Uri(url).Host); var size = new FileInfo(part).Length; File.Move(part, path, overwrite: true); Grew(size); return (ProxyOutcome.Cached, path, contentType); } finally { _downloads.Release(); if (File.Exists(part)) File.Delete(part); } } void Grew(long bytes) { if (Interlocked.Read(ref _bytes) < 0) Interlocked.CompareExchange(ref _bytes, Measure(), -1); if (Interlocked.Add(ref _bytes, bytes) > _options.CurrentValue.ProxyCacheBytes) _ = Task.Run(Trim); } public void Trim() { if (Interlocked.Exchange(ref _trimming, 1) == 1) return; try { Interlocked.Exchange(ref _bytes, TrimDirectory(_media.ProxyRoot, _options.CurrentValue.ProxyCacheBytes)); } finally { Interlocked.Exchange(ref _trimming, 0); } } public int Purge(string domain) { var directory = new DirectoryInfo(_media.ProxyRoot); if (string.IsNullOrEmpty(domain) || !directory.Exists) return 0; var purged = 0; foreach (var type in directory.EnumerateFiles("*.type", SearchOption.AllDirectories)) { try { var host = File.ReadLines(type.FullName).Skip(1).FirstOrDefault(); if (host == default || !(host.Equals(domain, StringComparison.OrdinalIgnoreCase) || host.EndsWith("." + domain, StringComparison.OrdinalIgnoreCase))) continue; var file = new FileInfo(type.FullName[..^".type".Length]); if (file.Exists) file.Delete(); type.Delete(); purged++; } catch (IOException) { } } Interlocked.Exchange(ref _bytes, -1); return purged; } long Measure() { var directory = new DirectoryInfo(_media.ProxyRoot); return directory.Exists ? directory.EnumerateFiles("*", SearchOption.AllDirectories).Where(f => f.Extension is not (".type" or ".part")).Sum(f => f.Length) : 0; } // deletes the oldest files until the cache is within its cap; what it holds afterwards public static long TrimDirectory(string root, long cap) { var directory = new DirectoryInfo(root); if (!directory.Exists) return 0; var files = directory.EnumerateFiles("*", SearchOption.AllDirectories).Where(f => f.Extension is not (".type" or ".part")).OrderBy(f => f.LastWriteTimeUtc).ToList(); var total = files.Sum(f => f.Length); foreach (var file in files) { if (total <= cap) break; try { total -= file.Length; file.Delete(); var type = new FileInfo(file.FullName + ".type"); if (type.Exists) type.Delete(); } catch (IOException) { } } return total; } (string Path, string TypePath) CachePaths(string url) { var name = Convert.ToHexStringLower(SHA256.HashData(Encoding.UTF8.GetBytes(url))); var path = System.IO.Path.Combine(_media.ProxyRoot, name[..2], name); return (path, path + ".type"); } string Sign(string url) => Base64Url(HMACSHA256.HashData(Key, Encoding.UTF8.GetBytes(url))[..16]); // the oldest key, so that two first uses at once agree on one static byte[] LoadKey() { var secret = DB.Default.Find().Sort(m => m.ID, Order.Ascending).ExecuteFirstAsync().GetAwaiter().GetResult(); if (secret == default) { DB.Default.SaveAsync(new MediaSecret { Key = Convert.ToBase64String(RandomNumberGenerator.GetBytes(32)) }).GetAwaiter().GetResult(); secret = DB.Default.Find().Sort(m => m.ID, Order.Ascending).ExecuteFirstAsync().GetAwaiter().GetResult(); } return Convert.FromBase64String(secret.Key); } static string Base64Url(byte[] bytes) => Convert.ToBase64String(bytes).TrimEnd('=').Replace('+', '-').Replace('/', '_'); static byte[] FromBase64Url(string value) { var padded = value.Replace('-', '+').Replace('_', '/'); return Convert.FromBase64String(padded + new string('=', (4 - padded.Length % 4) % 4)); } } public class MediaJanitor : BackgroundService { static readonly TimeSpan FirstPass = TimeSpan.FromMinutes(5); static readonly TimeSpan Interval = TimeSpan.FromHours(1); static readonly TimeSpan UnattachedLifetime = TimeSpan.FromDays(1); // a trashed file waits this long before it is deleted, already out of what /media/files serves public static readonly TimeSpan TrashGrace = TimeSpan.FromDays(1); readonly IMediaService _media; readonly IOptionsMonitor _options; readonly ILogger _logger; public MediaJanitor(IMediaService media, IOptionsMonitor options, ILogger logger) { _media = media; _options = options; _logger = logger; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { var wait = FirstPass; while (!stoppingToken.IsCancellationRequested) { try { await Task.Delay(wait, stoppingToken); wait = Interval; await Sweep(stoppingToken); } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { return; } catch (Exception ex) { _logger.LogWarning(ex, "{Service} pass failed", nameof(MediaJanitor)); } } } // one pass: // - uploads never posted for a day (and not waiting for a scheduled post, nor a profile picture) go to the trash; // - trashed files still served (a crash between the mark and the move) are moved out; // - trashed files past their grace are deleted, with their rows; // - the proxy cache is trimmed to its size, oldest first public async Task Sweep(CancellationToken token) { var cutoff = DateTime.UtcNow - UnattachedLifetime; var unattached = await _media.Trash(m => m.PostId == null && m.ScheduledStatusId == null && m.ProfileOfAvatarId == null && m.CreatedAt < cutoff, "never posted", token); var trashed = await DB.Default.Find().Match(m => m.TrashedAt != null).Limit(2000).ExecuteAsync(token); var purgeBefore = DateTime.UtcNow - TrashGrace; var purged = 0; foreach (var row in trashed) { if (row.TrashedAt < purgeBefore) { await _media.Purge(row, token); purged++; } else _media.Hide(row); } if (unattached > 0 || purged > 0) _logger.LogInformation("{Service}: {Unattached} uploads never posted trashed, {Purged} trashed files deleted", nameof(MediaJanitor), unattached, purged); TrimProxyCache(); } void TrimProxyCache() => MediaProxy.TrimDirectory(_media.ProxyRoot, _options.CurrentValue.ProxyCacheBytes); } }