fix: resolve 4 unhealthy services from fresh volume rebuild
- router-gateway: sync 10 missing jobengine routes to local config (prevent array merge bleed-through) - findings-ledger-web: add VulnExplorer tables to postgres-init bootstrap script - timeline-web: replace competing migration hosted service with standard AddStartupMigrations - graph-api: handle null PostgresGraphRepository gracefully, add graph schema to init - scheduler-web: add failure_signatures table to init bootstrap Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
master
@@ -1,5 +1,6 @@
|
||||
using StellaOps.Integrations.Contracts;
|
||||
using StellaOps.Integrations.Core;
|
||||
using System.Net;
|
||||
using System.Net.Http;
|
||||
using System.Net.Http.Headers;
|
||||
using System.Text.Json;
|
||||
@@ -47,7 +48,7 @@ public sealed class DockerRegistryConnectorPlugin : IIntegrationConnectorPlugin,
|
||||
try
|
||||
{
|
||||
// OCI Distribution Spec: GET /v2/ returns 200 {} when authenticated
|
||||
var response = await client.GetAsync("/v2/", cancellationToken);
|
||||
using var response = await SendGetAsync(client, config, "/v2/", cancellationToken);
|
||||
var duration = _timeProvider.GetUtcNow() - startTime;
|
||||
|
||||
if (response.IsSuccessStatusCode)
|
||||
@@ -99,7 +100,7 @@ public sealed class DockerRegistryConnectorPlugin : IIntegrationConnectorPlugin,
|
||||
try
|
||||
{
|
||||
// Check /v2/_catalog to verify registry is fully operational
|
||||
var response = await client.GetAsync("/v2/_catalog", cancellationToken);
|
||||
using var response = await SendGetAsync(client, config, "/v2/_catalog", cancellationToken);
|
||||
var duration = _timeProvider.GetUtcNow() - startTime;
|
||||
|
||||
if (response.IsSuccessStatusCode)
|
||||
@@ -147,18 +148,19 @@ public sealed class DockerRegistryConnectorPlugin : IIntegrationConnectorPlugin,
|
||||
|
||||
return IntegrationDiscoveryResourceTypes.Normalize(resourceType) switch
|
||||
{
|
||||
IntegrationDiscoveryResourceTypes.Repositories => await DiscoverRepositoriesAsync(client, filter, cancellationToken),
|
||||
IntegrationDiscoveryResourceTypes.Tags => await DiscoverTagsAsync(client, filter, cancellationToken),
|
||||
IntegrationDiscoveryResourceTypes.Repositories => await DiscoverRepositoriesAsync(client, config, filter, cancellationToken),
|
||||
IntegrationDiscoveryResourceTypes.Tags => await DiscoverTagsAsync(client, config, filter, cancellationToken),
|
||||
_ => []
|
||||
};
|
||||
}
|
||||
|
||||
private static async Task<IReadOnlyList<DiscoveredIntegrationResource>> DiscoverRepositoriesAsync(
|
||||
HttpClient client,
|
||||
IntegrationConfig config,
|
||||
IReadOnlyDictionary<string, string>? filter,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var response = await client.GetAsync("/v2/_catalog", cancellationToken);
|
||||
using var response = await SendGetAsync(client, config, "/v2/_catalog", cancellationToken);
|
||||
response.EnsureSuccessStatusCode();
|
||||
|
||||
var content = await response.Content.ReadAsStringAsync(cancellationToken);
|
||||
@@ -179,6 +181,7 @@ public sealed class DockerRegistryConnectorPlugin : IIntegrationConnectorPlugin,
|
||||
|
||||
private static async Task<IReadOnlyList<DiscoveredIntegrationResource>> DiscoverTagsAsync(
|
||||
HttpClient client,
|
||||
IntegrationConfig config,
|
||||
IReadOnlyDictionary<string, string>? filter,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
@@ -193,7 +196,7 @@ public sealed class DockerRegistryConnectorPlugin : IIntegrationConnectorPlugin,
|
||||
repository.Split('/', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries)
|
||||
.Select(Uri.EscapeDataString));
|
||||
|
||||
var response = await client.GetAsync($"/v2/{repositoryPath}/tags/list", cancellationToken);
|
||||
using var response = await SendGetAsync(client, config, $"/v2/{repositoryPath}/tags/list", cancellationToken);
|
||||
response.EnsureSuccessStatusCode();
|
||||
|
||||
var content = await response.Content.ReadAsStringAsync(cancellationToken);
|
||||
@@ -215,6 +218,125 @@ public sealed class DockerRegistryConnectorPlugin : IIntegrationConnectorPlugin,
|
||||
.ToList();
|
||||
}
|
||||
|
||||
private static async Task<HttpResponseMessage> SendGetAsync(
|
||||
HttpClient client,
|
||||
IntegrationConfig config,
|
||||
string requestUri,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var response = await client.GetAsync(requestUri, cancellationToken).ConfigureAwait(false);
|
||||
if (response.StatusCode != HttpStatusCode.Unauthorized || string.IsNullOrWhiteSpace(config.ResolvedSecret))
|
||||
{
|
||||
return response;
|
||||
}
|
||||
|
||||
var challenge = response.Headers.WwwAuthenticate
|
||||
.FirstOrDefault(header => string.Equals(header.Scheme, "Bearer", StringComparison.OrdinalIgnoreCase));
|
||||
if (challenge is null || !TryBuildTokenUri(challenge, out var tokenUri))
|
||||
{
|
||||
return response;
|
||||
}
|
||||
|
||||
var token = await RequestBearerTokenAsync(client, config.ResolvedSecret, tokenUri, cancellationToken).ConfigureAwait(false);
|
||||
if (string.IsNullOrWhiteSpace(token))
|
||||
{
|
||||
return response;
|
||||
}
|
||||
|
||||
response.Dispose();
|
||||
|
||||
using var retryRequest = new HttpRequestMessage(HttpMethod.Get, requestUri);
|
||||
retryRequest.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json"));
|
||||
retryRequest.Headers.Authorization = new AuthenticationHeaderValue("Bearer", token);
|
||||
return await client.SendAsync(retryRequest, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private static async Task<string?> RequestBearerTokenAsync(
|
||||
HttpClient client,
|
||||
string resolvedSecret,
|
||||
Uri tokenUri,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
using var request = new HttpRequestMessage(HttpMethod.Get, tokenUri);
|
||||
request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json"));
|
||||
ApplyResolvedSecret(request.Headers, resolvedSecret);
|
||||
|
||||
using var response = await client.SendAsync(request, cancellationToken).ConfigureAwait(false);
|
||||
if (!response.IsSuccessStatusCode)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
var content = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
|
||||
var tokenResponse = JsonSerializer.Deserialize<RegistryBearerTokenResponse>(content, JsonOptions);
|
||||
return tokenResponse?.Token ?? tokenResponse?.AccessToken;
|
||||
}
|
||||
|
||||
private static bool TryBuildTokenUri(AuthenticationHeaderValue challenge, out Uri tokenUri)
|
||||
{
|
||||
tokenUri = null!;
|
||||
|
||||
var parameters = ParseChallengeParameters(challenge.Parameter);
|
||||
if (!parameters.TryGetValue("realm", out var realm) || string.IsNullOrWhiteSpace(realm))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!Uri.TryCreate(realm, UriKind.Absolute, out var realmUri))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
var queryParts = new List<string>();
|
||||
if (parameters.TryGetValue("service", out var service) && !string.IsNullOrWhiteSpace(service))
|
||||
{
|
||||
queryParts.Add($"service={Uri.EscapeDataString(service)}");
|
||||
}
|
||||
|
||||
if (parameters.TryGetValue("scope", out var scope) && !string.IsNullOrWhiteSpace(scope))
|
||||
{
|
||||
queryParts.Add($"scope={Uri.EscapeDataString(scope)}");
|
||||
}
|
||||
|
||||
var builder = new UriBuilder(realmUri);
|
||||
var existingQuery = builder.Query.TrimStart('?');
|
||||
builder.Query = string.IsNullOrWhiteSpace(existingQuery)
|
||||
? string.Join("&", queryParts)
|
||||
: queryParts.Count == 0
|
||||
? existingQuery
|
||||
: $"{existingQuery}&{string.Join("&", queryParts)}";
|
||||
|
||||
tokenUri = builder.Uri;
|
||||
return true;
|
||||
}
|
||||
|
||||
private static Dictionary<string, string> ParseChallengeParameters(string? parameter)
|
||||
{
|
||||
var parameters = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
|
||||
if (string.IsNullOrWhiteSpace(parameter))
|
||||
{
|
||||
return parameters;
|
||||
}
|
||||
|
||||
foreach (var part in parameter.Split(',', StringSplitOptions.TrimEntries | StringSplitOptions.RemoveEmptyEntries))
|
||||
{
|
||||
var separatorIndex = part.IndexOf('=');
|
||||
if (separatorIndex <= 0)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
var key = part[..separatorIndex].Trim();
|
||||
var value = part[(separatorIndex + 1)..].Trim().Trim('"');
|
||||
if (!string.IsNullOrWhiteSpace(key))
|
||||
{
|
||||
parameters[key] = value;
|
||||
}
|
||||
}
|
||||
|
||||
return parameters;
|
||||
}
|
||||
|
||||
private static HttpClient CreateHttpClient(IntegrationConfig config)
|
||||
{
|
||||
var client = new HttpClient
|
||||
@@ -225,22 +347,29 @@ public sealed class DockerRegistryConnectorPlugin : IIntegrationConnectorPlugin,
|
||||
|
||||
client.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json"));
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(config.ResolvedSecret))
|
||||
{
|
||||
if (config.ResolvedSecret.Contains(':', StringComparison.Ordinal))
|
||||
{
|
||||
var credentials = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(config.ResolvedSecret));
|
||||
client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Basic", credentials);
|
||||
}
|
||||
else
|
||||
{
|
||||
client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", config.ResolvedSecret);
|
||||
}
|
||||
}
|
||||
ApplyResolvedSecret(client.DefaultRequestHeaders, config.ResolvedSecret);
|
||||
|
||||
return client;
|
||||
}
|
||||
|
||||
private static void ApplyResolvedSecret(HttpRequestHeaders headers, string? resolvedSecret)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(resolvedSecret))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (resolvedSecret.Contains(':', StringComparison.Ordinal))
|
||||
{
|
||||
var credentials = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(resolvedSecret));
|
||||
headers.Authorization = new AuthenticationHeaderValue("Basic", credentials);
|
||||
}
|
||||
else
|
||||
{
|
||||
headers.Authorization = new AuthenticationHeaderValue("Bearer", resolvedSecret);
|
||||
}
|
||||
}
|
||||
|
||||
private static readonly JsonSerializerOptions JsonOptions = new()
|
||||
{
|
||||
PropertyNameCaseInsensitive = true
|
||||
@@ -258,4 +387,10 @@ public sealed class DockerRegistryConnectorPlugin : IIntegrationConnectorPlugin,
|
||||
public string? Name { get; set; }
|
||||
public List<string>? Tags { get; set; }
|
||||
}
|
||||
|
||||
private sealed class RegistryBearerTokenResponse
|
||||
{
|
||||
public string? Token { get; set; }
|
||||
public string? AccessToken { get; set; }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -76,6 +76,33 @@ public sealed class DockerRegistryConnectorPluginTests
|
||||
request.Authorization);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task TestConnectionAsync_WithBearerChallenge_RetriesWithRegistryToken()
|
||||
{
|
||||
LoopbackHttpFixture? fixture = null;
|
||||
fixture = LoopbackHttpFixture.Start(3, request => request.Path switch
|
||||
{
|
||||
"/v2/" when request.Authorization is not null && request.Authorization.StartsWith("Basic ", StringComparison.Ordinal) =>
|
||||
HttpResponse.Challenge($"{fixture!.BaseUrl}/jwt/auth", "container_registry", "registry:catalog:*"),
|
||||
"/jwt/auth?service=container_registry&scope=registry%3Acatalog%3A%2A" when request.Authorization is not null && request.Authorization.StartsWith("Basic ", StringComparison.Ordinal) =>
|
||||
HttpResponse.Json("""{"token":"registry-jwt"}"""),
|
||||
"/v2/" when request.Authorization == "Bearer registry-jwt" =>
|
||||
HttpResponse.Json("{}"),
|
||||
_ => HttpResponse.Text("unexpected-path"),
|
||||
});
|
||||
|
||||
var plugin = new DockerRegistryConnectorPlugin();
|
||||
var result = await plugin.TestConnectionAsync(CreateConfig(fixture.BaseUrl, "root:registry-pat"));
|
||||
|
||||
var requests = await fixture.WaitForRequestsAsync();
|
||||
|
||||
Assert.True(result.Success);
|
||||
Assert.Equal(
|
||||
["/v2/", "/jwt/auth?service=container_registry&scope=registry%3Acatalog%3A%2A", "/v2/"],
|
||||
requests.Select(request => request.Path).ToArray());
|
||||
Assert.Equal("Bearer registry-jwt", requests[^1].Authorization);
|
||||
}
|
||||
|
||||
private static IntegrationConfig CreateConfig(string endpoint, string? resolvedSecret = null)
|
||||
{
|
||||
return new IntegrationConfig(
|
||||
@@ -91,23 +118,29 @@ public sealed class DockerRegistryConnectorPluginTests
|
||||
private sealed class LoopbackHttpFixture : IDisposable
|
||||
{
|
||||
private readonly TcpListener _listener;
|
||||
private readonly Task<HttpRequestData> _requestTask;
|
||||
private readonly Task<IReadOnlyList<HttpRequestData>> _requestTask;
|
||||
|
||||
private LoopbackHttpFixture(Func<string, HttpResponse> responder)
|
||||
private LoopbackHttpFixture(int expectedRequests, Func<HttpRequestData, HttpResponse> responder)
|
||||
{
|
||||
_listener = new TcpListener(IPAddress.Loopback, 0);
|
||||
_listener.Start();
|
||||
BaseUrl = $"http://127.0.0.1:{((IPEndPoint)_listener.LocalEndpoint).Port}";
|
||||
_requestTask = HandleSingleRequestAsync(responder);
|
||||
_requestTask = HandleRequestsAsync(expectedRequests, responder);
|
||||
}
|
||||
|
||||
public string BaseUrl { get; }
|
||||
|
||||
public static LoopbackHttpFixture Start(Func<string, HttpResponse> responder) => new(responder);
|
||||
public static LoopbackHttpFixture Start(Func<string, HttpResponse> responder) =>
|
||||
new(1, request => responder(request.Path));
|
||||
|
||||
public async Task<string> WaitForPathAsync() => (await _requestTask).Path;
|
||||
public static LoopbackHttpFixture Start(int expectedRequests, Func<HttpRequestData, HttpResponse> responder) =>
|
||||
new(expectedRequests, responder);
|
||||
|
||||
public Task<HttpRequestData> WaitForRequestAsync() => _requestTask;
|
||||
public async Task<string> WaitForPathAsync() => (await _requestTask)[0].Path;
|
||||
|
||||
public async Task<HttpRequestData> WaitForRequestAsync() => (await _requestTask)[0];
|
||||
|
||||
public Task<IReadOnlyList<HttpRequestData>> WaitForRequestsAsync() => _requestTask;
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
@@ -120,68 +153,96 @@ public sealed class DockerRegistryConnectorPluginTests
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<HttpRequestData> HandleSingleRequestAsync(Func<string, HttpResponse> responder)
|
||||
private async Task<IReadOnlyList<HttpRequestData>> HandleRequestsAsync(int expectedRequests, Func<HttpRequestData, HttpResponse> responder)
|
||||
{
|
||||
using var client = await _listener.AcceptTcpClientAsync();
|
||||
using var stream = client.GetStream();
|
||||
using var reader = new StreamReader(
|
||||
stream,
|
||||
Encoding.ASCII,
|
||||
detectEncodingFromByteOrderMarks: false,
|
||||
bufferSize: 1024,
|
||||
leaveOpen: true);
|
||||
var requests = new List<HttpRequestData>(expectedRequests);
|
||||
|
||||
var requestLine = await reader.ReadLineAsync();
|
||||
if (string.IsNullOrWhiteSpace(requestLine))
|
||||
for (var i = 0; i < expectedRequests; i++)
|
||||
{
|
||||
throw new InvalidOperationException("Did not receive an HTTP request line.");
|
||||
}
|
||||
using var client = await _listener.AcceptTcpClientAsync();
|
||||
using var stream = client.GetStream();
|
||||
using var reader = new StreamReader(
|
||||
stream,
|
||||
Encoding.ASCII,
|
||||
detectEncodingFromByteOrderMarks: false,
|
||||
bufferSize: 1024,
|
||||
leaveOpen: true);
|
||||
|
||||
var requestParts = requestLine.Split(' ', StringSplitOptions.RemoveEmptyEntries);
|
||||
if (requestParts.Length < 2)
|
||||
{
|
||||
throw new InvalidOperationException($"Unexpected HTTP request line: {requestLine}");
|
||||
}
|
||||
|
||||
string? authorization = null;
|
||||
while (true)
|
||||
{
|
||||
var headerLine = await reader.ReadLineAsync();
|
||||
if (string.IsNullOrEmpty(headerLine))
|
||||
var requestLine = await reader.ReadLineAsync();
|
||||
if (string.IsNullOrWhiteSpace(requestLine))
|
||||
{
|
||||
break;
|
||||
throw new InvalidOperationException("Did not receive an HTTP request line.");
|
||||
}
|
||||
|
||||
if (headerLine.StartsWith("Authorization:", StringComparison.OrdinalIgnoreCase))
|
||||
var requestParts = requestLine.Split(' ', StringSplitOptions.RemoveEmptyEntries);
|
||||
if (requestParts.Length < 2)
|
||||
{
|
||||
authorization = headerLine["Authorization:".Length..].Trim();
|
||||
throw new InvalidOperationException($"Unexpected HTTP request line: {requestLine}");
|
||||
}
|
||||
|
||||
string? authorization = null;
|
||||
while (true)
|
||||
{
|
||||
var headerLine = await reader.ReadLineAsync();
|
||||
if (string.IsNullOrEmpty(headerLine))
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
if (headerLine.StartsWith("Authorization:", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
authorization = headerLine["Authorization:".Length..].Trim();
|
||||
}
|
||||
}
|
||||
|
||||
var request = new HttpRequestData(requestParts[1], authorization);
|
||||
requests.Add(request);
|
||||
|
||||
var response = responder(request);
|
||||
var payload = Encoding.UTF8.GetBytes(response.Body);
|
||||
IEnumerable<KeyValuePair<string, string>> responseHeaders = response.Headers is null
|
||||
? Array.Empty<KeyValuePair<string, string>>()
|
||||
: response.Headers;
|
||||
var responseText =
|
||||
$"HTTP/1.1 {response.StatusCode} {response.ReasonPhrase}\r\n" +
|
||||
$"Content-Type: {response.ContentType}\r\n" +
|
||||
$"Content-Length: {payload.Length}\r\n" +
|
||||
string.Concat(responseHeaders.Select(header => $"{header.Key}: {header.Value}\r\n")) +
|
||||
"Connection: close\r\n" +
|
||||
"\r\n";
|
||||
|
||||
var headerBytes = Encoding.ASCII.GetBytes(responseText);
|
||||
await stream.WriteAsync(headerBytes);
|
||||
await stream.WriteAsync(payload);
|
||||
await stream.FlushAsync();
|
||||
}
|
||||
|
||||
var requestPath = requestParts[1];
|
||||
var response = responder(requestPath);
|
||||
var payload = Encoding.UTF8.GetBytes(response.Body);
|
||||
var responseText =
|
||||
$"HTTP/1.1 {response.StatusCode} {response.ReasonPhrase}\r\n" +
|
||||
$"Content-Type: {response.ContentType}\r\n" +
|
||||
$"Content-Length: {payload.Length}\r\n" +
|
||||
"Connection: close\r\n" +
|
||||
"\r\n";
|
||||
|
||||
var headerBytes = Encoding.ASCII.GetBytes(responseText);
|
||||
await stream.WriteAsync(headerBytes);
|
||||
await stream.WriteAsync(payload);
|
||||
await stream.FlushAsync();
|
||||
return new HttpRequestData(requestPath, authorization);
|
||||
return requests;
|
||||
}
|
||||
}
|
||||
|
||||
private sealed record HttpRequestData(string Path, string? Authorization);
|
||||
|
||||
private sealed record HttpResponse(int StatusCode, string ReasonPhrase, string ContentType, string Body)
|
||||
private sealed record HttpResponse(
|
||||
int StatusCode,
|
||||
string ReasonPhrase,
|
||||
string ContentType,
|
||||
string Body,
|
||||
IReadOnlyDictionary<string, string>? Headers = null)
|
||||
{
|
||||
public static HttpResponse Json(string body) => new(200, "OK", "application/json", body);
|
||||
|
||||
public static HttpResponse Text(string body) => new(200, "OK", "text/plain", body);
|
||||
|
||||
public static HttpResponse Challenge(string realm, string service, string scope) =>
|
||||
new(
|
||||
401,
|
||||
"Unauthorized",
|
||||
"text/plain",
|
||||
string.Empty,
|
||||
new Dictionary<string, string>
|
||||
{
|
||||
["WWW-Authenticate"] = $"Bearer realm=\"{realm}\",service=\"{service}\",scope=\"{scope}\""
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user