using System.Net; using System.Net.Http.Headers; namespace Quorate.Oci; internal sealed class OciPublisher(RegistryTransport transport, RepositoryName repository) { private static readonly string[] ManifestTypes = [ "application/vnd.docker.distribution.manifest.v2+json", "application/vnd.docker.distribution.manifest.list.v2+json", "application/vnd.oci.image.manifest.v1+json", "application/vnd.oci.image.index.v1+json" ]; private static readonly TimeSpan UploadCancelTimeout = TimeSpan.FromSeconds(5); public async Task PushAsync( OciPublicationPlan plan, OciTag target, CancellationToken cancellationToken) { var parallel = new ParallelOptions { CancellationToken = cancellationToken, MaxDegreeOfParallelism = 4 }; await Parallel.ForEachAsync(plan.Blobs, parallel, async (blob, token) => { await VerifyBlobAsync(blob, token).ConfigureAwait(false); if (!await ExistsAsync(blob, token).ConfigureAwait(false)) { await UploadAsync(blob, token).ConfigureAwait(false); } }).ConfigureAwait(false); foreach (OciManifest manifest in plan.Manifests) { await PutManifestAsync( manifest.Digest.ToString(), manifest.MediaType, manifest.Content, manifest.Digest, cancellationToken).ConfigureAwait(false); } await PutManifestAsync( target.ToString(), plan.RootMediaType, plan.RootContent, plan.RootDigest, cancellationToken).ConfigureAwait(false); OciDigest published = await GetDigestAsync(target, cancellationToken).ConfigureAwait(false); if (published != plan.RootDigest) { throw new InvalidDataException($"Published digest {published} does not match {plan.RootDigest}."); } return published; } private async Task ExistsAsync(OciBlob blob, CancellationToken cancellationToken) { using HttpResponseMessage response = await transport.SendAsync( repository, () => RegistryTransport.Request(HttpMethod.Head, transport.Api(repository, $"blobs/{blob.Digest}")), cancellationToken, HttpStatusCode.NotFound).ConfigureAwait(false); if (response.StatusCode == HttpStatusCode.NotFound) { return false; } OciDigest actual = HeaderDigest(response); long? size = response.Content.Headers.ContentLength; if (actual != blob.Digest || size != blob.Size) { throw new InvalidDataException($"Existing blob {blob.Digest} has invalid metadata."); } return true; } private async Task UploadAsync(OciBlob blob, CancellationToken cancellationToken) { using HttpResponseMessage start = await transport.SendAsync( repository, () => RegistryTransport.Request(HttpMethod.Post, transport.Api(repository, "blobs/uploads/")), cancellationToken).ConfigureAwait(false); Uri location = RegistryTransport.ResolveLocation(start); try { using HttpResponseMessage completed = await transport.SendAsync( repository, () => { HttpRequestMessage request = RegistryTransport.Request( HttpMethod.Put, AddDigest(location, blob.Digest)); request.Content = new StreamContent(new FileStream( blob.Path, FileMode.Open, FileAccess.Read, FileShare.Read, 131072, FileOptions.Asynchronous | FileOptions.SequentialScan)); request.Content.Headers.ContentType = new MediaTypeHeaderValue("application/octet-stream"); return request; }, cancellationToken).ConfigureAwait(false); OciDigest actual = HeaderDigest(completed); if (actual != blob.Digest) { throw new InvalidDataException($"Uploaded digest {actual} does not match {blob.Digest}."); } } catch (Exception) { await CancelUploadAsync(location).ConfigureAwait(false); throw; } } private async Task CancelUploadAsync(Uri location) { using var timeout = new CancellationTokenSource(UploadCancelTimeout); try { using HttpResponseMessage response = await transport.SendAsync( repository, () => RegistryTransport.Request(HttpMethod.Delete, location), timeout.Token, HttpStatusCode.NotFound).ConfigureAwait(false); } catch (Exception error) when (error is HttpRequestException or InvalidDataException or OperationCanceledException) { } } private async Task PutManifestAsync( string reference, string mediaType, byte[] content, OciDigest expected, CancellationToken cancellationToken) { using HttpResponseMessage response = await transport.SendAsync( repository, () => { HttpRequestMessage request = RegistryTransport.Request( HttpMethod.Put, transport.Api(repository, $"manifests/{reference}")); request.Content = new ByteArrayContent(content); request.Content.Headers.ContentType = new MediaTypeHeaderValue(mediaType); return request; }, cancellationToken).ConfigureAwait(false); OciDigest actual = HeaderDigest(response); if (actual != expected) { throw new InvalidDataException($"Published digest {actual} does not match {expected}."); } } private async Task GetDigestAsync(OciTag reference, CancellationToken cancellationToken) { using HttpResponseMessage response = await transport.SendAsync( repository, () => { HttpRequestMessage request = RegistryTransport.Request( HttpMethod.Head, transport.Api(repository, $"manifests/{reference}")); foreach (string mediaType in ManifestTypes) { request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue(mediaType)); } return request; }, cancellationToken).ConfigureAwait(false); return HeaderDigest(response); } private static async Task VerifyBlobAsync(OciBlob blob, CancellationToken cancellationToken) { FileInfo info = new(blob.Path); if (!info.Exists || info.Length != blob.Size) { throw new InvalidDataException($"OCI blob {blob.Digest} has an invalid size."); } OciDigest actual = await OciDigest.ComputeFileAsync(blob.Path, cancellationToken).ConfigureAwait(false); if (actual != blob.Digest) { throw new InvalidDataException($"Blob digest {actual} does not match {blob.Digest}."); } } private static Uri AddDigest(Uri location, OciDigest digest) { var builder = new UriBuilder(location); string pair = "digest=" + Uri.EscapeDataString(digest.ToString()); builder.Query = string.IsNullOrEmpty(builder.Query) ? pair : builder.Query.TrimStart('?') + "&" + pair; return builder.Uri; } private static OciDigest HeaderDigest(HttpResponseMessage response) => response.Headers.TryGetValues("Docker-Content-Digest", out IEnumerable? values) ? OciDigest.Parse(values.Single()) : throw new InvalidDataException("Registry response lacks Docker-Content-Digest."); }