fix(a2a): dedicated guarded HTTP clients, context-checked continuations, request-scoped card URL
This commit is contained in:
@@ -15,7 +15,7 @@ namespace Volcanic.AgenticCore.A2A.Server.Agent;
|
||||
/// before its first event and stays live until its final status is written, so restart recovery never touches it.
|
||||
/// </summary>
|
||||
internal sealed class A2AAgentHandler(
|
||||
A2ACaller caller, A2ASessionPool pool, IOptions<A2AHostOptions> options, ISecretRedactor redactor, HttpClient http,
|
||||
A2ACaller caller, A2ASessionPool pool, IOptions<A2AHostOptions> options, ISecretRedactor redactor, A2AFileDownloader downloader,
|
||||
TimeProvider time, ILogger<A2AAgentHandler> logger) : IAgentHandler
|
||||
{
|
||||
/// <summary>Bound on the final status write: after a client disconnect the bounded event queue may never be read.</summary>
|
||||
@@ -82,7 +82,7 @@ internal sealed class A2AAgentHandler(
|
||||
string prompt;
|
||||
try
|
||||
{
|
||||
prompt = await A2APrompt.BuildAsync(caller, context.Message, options.Value.MaxFileBytes, http, time, token).ConfigureAwait(false);
|
||||
prompt = await A2APrompt.BuildAsync(caller, context.Message, options.Value.MaxFileBytes, downloader.Http, time, token).ConfigureAwait(false);
|
||||
}
|
||||
catch (A2APromptException ex)
|
||||
{
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
namespace Volcanic.AgenticCore.A2A.Server.Agent;
|
||||
|
||||
/// <summary>
|
||||
/// The HTTP client for file-part URL downloads. A dedicated holder, so the A2A server never registers a bare
|
||||
/// <see cref="HttpClient"/> that other services (provider extensions) would pick up.
|
||||
/// </summary>
|
||||
internal sealed class A2AFileDownloader(HttpClient http)
|
||||
{
|
||||
public HttpClient Http => http;
|
||||
}
|
||||
@@ -18,6 +18,8 @@ internal static class A2AAuth
|
||||
return auth.StartsWith("Bearer ", StringComparison.OrdinalIgnoreCase) ? auth["Bearer ".Length..].Trim() : null;
|
||||
}
|
||||
|
||||
internal static void SetCaller(HttpContext context, A2ACaller caller) => context.Items[ItemKey] = caller;
|
||||
|
||||
public static A2ACaller CallerOf(HttpContext context) =>
|
||||
context.Items[ItemKey] as A2ACaller ?? throw new InvalidOperationException("The A2A request was not authenticated.");
|
||||
|
||||
@@ -35,7 +37,7 @@ internal static class A2AAuth
|
||||
string.Join(", ", matches.Select(m => m.OwnerId)));
|
||||
return Results.Unauthorized();
|
||||
}
|
||||
http.Items[ItemKey] = matches[0];
|
||||
SetCaller(http, matches[0]);
|
||||
return await next(invocation).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -79,6 +79,13 @@ internal sealed class A2ACardProvider(IOptions<A2AHostOptions> options, ILogger<
|
||||
? Directory.EnumerateFiles(workspace).FirstOrDefault(f => string.Equals(Path.GetFileName(f), "PERSONA.md", StringComparison.OrdinalIgnoreCase))
|
||||
: null;
|
||||
if (persona is null) return null;
|
||||
return File.ReadLines(persona).Select(l => l.Trim().TrimStart('#').Trim()).FirstOrDefault(l => l.Length > 0);
|
||||
try
|
||||
{
|
||||
return File.ReadLines(persona).Select(l => l.Trim().TrimStart('#').Trim()).FirstOrDefault(l => l.Length > 0);
|
||||
}
|
||||
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,19 +3,15 @@ using Microsoft.AspNetCore.Http;
|
||||
|
||||
namespace Volcanic.AgenticCore.A2A.Server.Hosting;
|
||||
|
||||
/// <summary>The handler the SDK endpoints call: forwards each request to the authenticated caller's server.</summary>
|
||||
internal sealed class A2ARequestRouter(IHttpContextAccessor http, A2AServerCache servers) : IA2ARequestHandler
|
||||
/// <summary>
|
||||
/// The handler the SDK endpoints call: forwards each request to the authenticated caller's server. The extended card
|
||||
/// is answered here, with the current request's base URL.
|
||||
/// </summary>
|
||||
internal sealed class A2ARequestRouter(IHttpContextAccessor http, A2AServerCache servers, A2ACardProvider cards) : IA2ARequestHandler
|
||||
{
|
||||
private VolcanicA2AServer Server
|
||||
{
|
||||
get
|
||||
{
|
||||
var context = http.HttpContext ?? throw new InvalidOperationException("A2A requests need an HTTP context.");
|
||||
var server = servers.For(A2AAuth.CallerOf(context));
|
||||
server.RequestBaseUrl = $"{context.Request.Scheme}://{context.Request.Host}";
|
||||
return server;
|
||||
}
|
||||
}
|
||||
private HttpContext Context => http.HttpContext ?? throw new InvalidOperationException("A2A requests need an HTTP context.");
|
||||
|
||||
private VolcanicA2AServer Server => servers.For(A2AAuth.CallerOf(Context));
|
||||
|
||||
public Task<SendMessageResponse> SendMessageAsync(SendMessageRequest request, CancellationToken cancellationToken = default) => Server.SendMessageAsync(request, cancellationToken);
|
||||
public IAsyncEnumerable<StreamResponse> SendStreamingMessageAsync(SendMessageRequest request, CancellationToken cancellationToken = default) => Server.SendStreamingMessageAsync(request, cancellationToken);
|
||||
@@ -27,5 +23,9 @@ internal sealed class A2ARequestRouter(IHttpContextAccessor http, A2AServerCache
|
||||
public Task<TaskPushNotificationConfig> GetTaskPushNotificationConfigAsync(GetTaskPushNotificationConfigRequest request, CancellationToken cancellationToken = default) => Server.GetTaskPushNotificationConfigAsync(request, cancellationToken);
|
||||
public Task<ListTaskPushNotificationConfigResponse> ListTaskPushNotificationConfigAsync(ListTaskPushNotificationConfigRequest request, CancellationToken cancellationToken = default) => Server.ListTaskPushNotificationConfigAsync(request, cancellationToken);
|
||||
public Task DeleteTaskPushNotificationConfigAsync(DeleteTaskPushNotificationConfigRequest request, CancellationToken cancellationToken = default) => Server.DeleteTaskPushNotificationConfigAsync(request, cancellationToken);
|
||||
public Task<AgentCard> GetExtendedAgentCardAsync(GetExtendedAgentCardRequest request, CancellationToken cancellationToken = default) => Server.GetExtendedAgentCardAsync(request, cancellationToken);
|
||||
public Task<AgentCard> GetExtendedAgentCardAsync(GetExtendedAgentCardRequest request, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var context = Context;
|
||||
return Task.FromResult(cards.Extended(A2AAuth.CallerOf(context), $"{context.Request.Scheme}://{context.Request.Host}"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -8,19 +8,21 @@ namespace Volcanic.AgenticCore.A2A.Server.Hosting;
|
||||
|
||||
/// <summary>
|
||||
/// One <see cref="VolcanicA2AServer"/> per caller identity (workspace + name). The SDK server keeps background runs and
|
||||
/// cancellation state, so it must outlive requests. Caller configuration is still read per request (auth filter).
|
||||
/// cancellation state, so it must outlive requests. The server captures the caller record seen on first use; the auth
|
||||
/// filter still reads callers.json on every request, so a removed key stops working at once.
|
||||
/// The pool is resolved in the constructor so the container disposes this cache (and its servers' runs) before the pool.
|
||||
/// </summary>
|
||||
internal sealed class A2AServerCache(IServiceProvider services) : IAsyncDisposable
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, Lazy<VolcanicA2AServer>> _servers = new();
|
||||
private readonly A2ASessionPool _pool = services.GetRequiredService<A2ASessionPool>();
|
||||
|
||||
public VolcanicA2AServer For(A2ACaller caller) =>
|
||||
_servers.GetOrAdd(caller.Id, _ => new Lazy<VolcanicA2AServer>(() => Create(caller))).Value;
|
||||
|
||||
private VolcanicA2AServer Create(A2ACaller caller)
|
||||
{
|
||||
var pool = services.GetRequiredService<A2ASessionPool>();
|
||||
var store = ActivatorUtilities.CreateInstance<WorkspaceTaskStore>(services, caller, (IA2ALiveRuns)pool);
|
||||
var store = ActivatorUtilities.CreateInstance<WorkspaceTaskStore>(services, caller, (IA2ALiveRuns)_pool);
|
||||
var handler = ActivatorUtilities.CreateInstance<A2AAgentHandler>(services, caller);
|
||||
return ActivatorUtilities.CreateInstance<VolcanicA2AServer>(services, caller, handler, store);
|
||||
}
|
||||
|
||||
@@ -24,9 +24,11 @@ public static class A2AServerExtensions
|
||||
if (configure is not null) services.Configure(configure);
|
||||
services.AddHttpContextAccessor();
|
||||
services.TryAddSingleton(TimeProvider.System);
|
||||
services.TryAddSingleton(new HttpClient { Timeout = TimeSpan.FromMinutes(2) }); // file-part downloads
|
||||
services.TryAddSingleton<A2APushSender>(sp => new HttpA2APushSender(new HttpClient(), sp.GetRequiredService<TimeProvider>(),
|
||||
sp.GetRequiredService<ILogger<HttpA2APushSender>>()));
|
||||
// Dedicated, SSRF-guarded clients (no redirects, private addresses refused at connect time). Never register a bare
|
||||
// HttpClient: core extensions built through ActivatorUtilities would pick it up.
|
||||
services.TryAddSingleton(sp => new A2AFileDownloader(new HttpClient(GuardedHandler(sp)) { Timeout = TimeSpan.FromMinutes(2) }));
|
||||
services.TryAddSingleton<A2APushSender>(sp => new HttpA2APushSender(new HttpClient(GuardedHandler(sp)),
|
||||
sp.GetRequiredService<TimeProvider>(), sp.GetRequiredService<ILogger<HttpA2APushSender>>()));
|
||||
services.TryAddSingleton(sp => new PushUrlGuard(sp.GetRequiredService<IOptions<A2AHostOptions>>()));
|
||||
services.TryAddSingleton<A2ASessionPool>();
|
||||
services.TryAddSingleton<A2ACardProvider>();
|
||||
@@ -35,6 +37,12 @@ public static class A2AServerExtensions
|
||||
return services;
|
||||
}
|
||||
|
||||
private static SocketsHttpHandler GuardedHandler(IServiceProvider sp)
|
||||
{
|
||||
var options = sp.GetRequiredService<IOptions<A2AHostOptions>>();
|
||||
return PushUrlGuard.CreateGuardedHandler(() => options.Value.AllowPrivatePushUrls);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Maps <c>/.well-known/agent-card.json</c> (public), JSON-RPC at <paramref name="path"/> and REST at
|
||||
/// <c>{path}/v1</c> (both behind the key filter), and the push receiver at <c>{path}/push/{watchId}</c>.
|
||||
|
||||
@@ -17,9 +17,6 @@ internal sealed class VolcanicA2AServer(
|
||||
PushUrlGuard guard, IOptions<A2AHostOptions> options, ILogger<A2AServer> logger)
|
||||
: A2AServer(handler, store, new ChannelEventNotifier(), logger, new A2AServerOptions())
|
||||
{
|
||||
/// <summary>Base URL of the current request, for the extended card when PublicUrl is unset.</summary>
|
||||
public string? RequestBaseUrl { get; set; }
|
||||
|
||||
public override Task<SendMessageResponse> SendMessageAsync(SendMessageRequest request, CancellationToken cancellationToken = default)
|
||||
{
|
||||
PreCheck(request);
|
||||
@@ -42,9 +39,16 @@ internal sealed class VolcanicA2AServer(
|
||||
if (part.Url is { } url && !url.StartsWith("https://", StringComparison.OrdinalIgnoreCase))
|
||||
throw new A2AException("File part URLs must be https.", A2AErrorCode.InvalidParams);
|
||||
}
|
||||
if (message.TaskId is { Length: > 0 } taskId && store.Peek(taskId) is { } task && task.Status.State != TaskState.InputRequired && !task.Status.State.IsTerminal())
|
||||
throw new A2AException($"Task '{taskId}' is not waiting for input.", A2AErrorCode.UnsupportedOperation);
|
||||
if (message.ContextId is { Length: > 0 } contextId && pool.IsBusy(A2ASessionPool.ScopeFor(caller, contextId)))
|
||||
var task = message.TaskId is { Length: > 0 } taskId ? store.Peek(taskId) : null;
|
||||
if (task is not null)
|
||||
{
|
||||
if (message.ContextId is { Length: > 0 } given && given != task.ContextId)
|
||||
throw new A2AException("The task belongs to another context.", A2AErrorCode.InvalidParams);
|
||||
if (task.Status.State != TaskState.InputRequired && !task.Status.State.IsTerminal())
|
||||
throw new A2AException($"Task '{task.Id}' is not waiting for input.", A2AErrorCode.UnsupportedOperation);
|
||||
}
|
||||
var contextId = message.ContextId is { Length: > 0 } c ? c : task?.ContextId;
|
||||
if (contextId is { Length: > 0 } && pool.IsBusy(A2ASessionPool.ScopeFor(caller, contextId)))
|
||||
throw new A2AException("Context busy: retry when the current task ends.", A2AErrorCode.UnsupportedOperation);
|
||||
}
|
||||
|
||||
@@ -54,6 +58,8 @@ internal sealed class VolcanicA2AServer(
|
||||
var config = request.Config ?? throw new A2AException("A push notification config is required.", A2AErrorCode.InvalidParams);
|
||||
if (await guard.ProblemAsync(config.Url, cancellationToken).ConfigureAwait(false) is { } problem)
|
||||
throw new A2AException(problem, A2AErrorCode.InvalidParams);
|
||||
if (PushUrlGuard.HeaderProblem(config) is { } headerProblem)
|
||||
throw new A2AException(headerProblem, A2AErrorCode.InvalidParams);
|
||||
// The SDK passes "" (not null) when the client sent no id.
|
||||
var id = !string.IsNullOrEmpty(request.ConfigId) ? request.ConfigId
|
||||
: !string.IsNullOrEmpty(config.Id) ? config.Id
|
||||
@@ -66,21 +72,34 @@ internal sealed class VolcanicA2AServer(
|
||||
}
|
||||
|
||||
public override Task<TaskPushNotificationConfig> GetTaskPushNotificationConfigAsync(
|
||||
GetTaskPushNotificationConfigRequest request, CancellationToken cancellationToken = default) =>
|
||||
Task.FromResult(store.PushConfigs(request.TaskId).FirstOrDefault(c => c.Id == request.Id)
|
||||
GetTaskPushNotificationConfigRequest request, CancellationToken cancellationToken = default)
|
||||
{
|
||||
// A null request is the SDK's capability probe: answer "nothing" instead of failing.
|
||||
if (request is null) return Task.FromResult<TaskPushNotificationConfig>(null!);
|
||||
return Task.FromResult(store.PushConfigs(request.TaskId).FirstOrDefault(c => c.Id == request.Id)
|
||||
?? throw new A2AException($"Push config '{request.Id}' not found.", A2AErrorCode.TaskNotFound));
|
||||
}
|
||||
|
||||
public override Task<ListTaskPushNotificationConfigResponse> ListTaskPushNotificationConfigAsync(
|
||||
ListTaskPushNotificationConfigRequest request, CancellationToken cancellationToken = default) =>
|
||||
Task.FromResult(new ListTaskPushNotificationConfigResponse { Configs = [.. store.PushConfigs(request.TaskId)], NextPageToken = "" });
|
||||
ListTaskPushNotificationConfigRequest request, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (store.Peek(request.TaskId) is null)
|
||||
throw new A2AException($"Task '{request.TaskId}' not found.", A2AErrorCode.TaskNotFound);
|
||||
return Task.FromResult(new ListTaskPushNotificationConfigResponse { Configs = [.. store.PushConfigs(request.TaskId)], NextPageToken = "" });
|
||||
}
|
||||
|
||||
public override Task DeleteTaskPushNotificationConfigAsync(
|
||||
DeleteTaskPushNotificationConfigRequest request, CancellationToken cancellationToken = default)
|
||||
{
|
||||
store.UpdatePushConfigs(request.TaskId, list => list.RemoveAll(c => c.Id == request.Id));
|
||||
if (!store.UpdatePushConfigs(request.TaskId, list => list.RemoveAll(c => c.Id == request.Id)))
|
||||
throw new A2AException($"Task '{request.TaskId}' not found.", A2AErrorCode.TaskNotFound);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The extended card without a request base URL (PublicUrl or localhost). Over HTTP the router answers this itself
|
||||
/// with the current request's base URL.
|
||||
/// </summary>
|
||||
public override Task<AgentCard> GetExtendedAgentCardAsync(GetExtendedAgentCardRequest request, CancellationToken cancellationToken = default) =>
|
||||
Task.FromResult(cards.Extended(caller, RequestBaseUrl));
|
||||
Task.FromResult(cards.Extended(caller, null));
|
||||
}
|
||||
|
||||
@@ -44,12 +44,17 @@ internal sealed class HttpA2APushSender(HttpClient http, TimeProvider time, Micr
|
||||
using var response = await http.SendAsync(request, timeout.Token).ConfigureAwait(false);
|
||||
if (response.IsSuccessStatusCode) return true;
|
||||
}
|
||||
catch (Exception ex) when (ex is HttpRequestException or TaskCanceledException) { }
|
||||
catch (Exception)
|
||||
{
|
||||
// Any failure (network, timeout, a bad stored URL or header) counts as a failed attempt: nothing may
|
||||
// escape the fire-and-forget Task.Run.
|
||||
}
|
||||
if (attempt >= Waits.Length) break;
|
||||
await Task.Delay(Waits[attempt], time).ConfigureAwait(false);
|
||||
}
|
||||
Microsoft.Extensions.Logging.LoggerExtensions.LogWarning(logger,
|
||||
"A2A: push for task {TaskId} to {Host} failed after 3 attempts", taskId, new Uri(target.Url).Host);
|
||||
"A2A: push for task {TaskId} to {Host} failed after 3 attempts", taskId,
|
||||
Uri.TryCreate(target.Url, UriKind.Absolute, out var uri) ? uri.Host : "(invalid URL)");
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,10 +1,15 @@
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using A2A;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
namespace Volcanic.AgenticCore.A2A.Server.Push;
|
||||
|
||||
/// <summary>Webhook URLs registered by callers: https only, and never loopback/private/link-local/CGNAT (SSRF).</summary>
|
||||
/// <summary>
|
||||
/// Outbound URLs chosen by callers (push webhooks, file-part downloads): https only, and never
|
||||
/// loopback/private/link-local/CGNAT/benchmark/multicast/reserved addresses (SSRF), unless
|
||||
/// <see cref="A2AHostOptions.AllowPrivatePushUrls"/>.
|
||||
/// </summary>
|
||||
internal sealed class PushUrlGuard(IOptions<A2AHostOptions> options, Func<string, CancellationToken, Task<IPAddress[]>>? resolve = null)
|
||||
{
|
||||
private readonly Func<string, CancellationToken, Task<IPAddress[]>> _resolve = resolve ?? Dns.GetHostAddressesAsync;
|
||||
@@ -29,17 +34,72 @@ internal sealed class PushUrlGuard(IOptions<A2AHostOptions> options, Func<string
|
||||
return addresses.Any(IsPrivate) ? "the push URL points to a private or loopback address" : null;
|
||||
}
|
||||
|
||||
/// <summary>Null when the config's token and auth scheme are safe to send as headers, else the reason.</summary>
|
||||
public static string? HeaderProblem(PushNotificationConfig config)
|
||||
{
|
||||
if (config.Token is { } token && token.AsSpan().IndexOfAny('\r', '\n', '\0') >= 0)
|
||||
return "the push token must not contain line breaks";
|
||||
if (config.Authentication is { } auth)
|
||||
{
|
||||
if (auth.Scheme.Length == 0 || !auth.Scheme.All(IsTokenChar)) return "the push auth scheme must be a single token";
|
||||
if (auth.Credentials is { } c && c.AsSpan().IndexOfAny('\r', '\n', '\0') >= 0)
|
||||
return "the push credentials must not contain line breaks";
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// RFC 9110 tchar.
|
||||
private static bool IsTokenChar(char c) => char.IsAsciiLetterOrDigit(c) || "!#$%&'*+-.^_`|~".Contains(c);
|
||||
|
||||
internal static bool IsPrivate(IPAddress address)
|
||||
{
|
||||
if (IPAddress.IsLoopback(address)) return true;
|
||||
if (address.IsIPv4MappedToIPv6) address = address.MapToIPv4();
|
||||
if (address.AddressFamily == AddressFamily.InterNetworkV6)
|
||||
return address.IsIPv6LinkLocal || address.IsIPv6SiteLocal || address.IsIPv6UniqueLocal || address.Equals(IPAddress.IPv6Any);
|
||||
return address.IsIPv6LinkLocal || address.IsIPv6SiteLocal || address.IsIPv6UniqueLocal || address.IsIPv6Multicast
|
||||
|| address.Equals(IPAddress.IPv6Any);
|
||||
var b = address.GetAddressBytes();
|
||||
return b[0] == 10 || b[0] == 127 || b[0] == 0
|
||||
|| b[0] >= 224 // multicast 224/4, reserved 240/4, broadcast
|
||||
|| (b[0] == 172 && b[1] >= 16 && b[1] <= 31)
|
||||
|| (b[0] == 192 && b[1] == 168)
|
||||
|| (b[0] == 169 && b[1] == 254)
|
||||
|| (b[0] == 100 && b[1] >= 64 && b[1] <= 127);
|
||||
|| (b[0] == 100 && b[1] >= 64 && b[1] <= 127)
|
||||
|| (b[0] == 198 && b[1] is 18 or 19);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A handler for caller-chosen URLs: no redirects, and every connection goes only to addresses checked with
|
||||
/// <see cref="IsPrivate"/> at connect time (so DNS rebinding can't reach internal hosts), unless
|
||||
/// <paramref name="allowPrivate"/> returns true.
|
||||
/// </summary>
|
||||
public static SocketsHttpHandler CreateGuardedHandler(Func<bool> allowPrivate,
|
||||
Func<string, CancellationToken, Task<IPAddress[]>>? resolve = null)
|
||||
{
|
||||
var resolver = resolve ?? Dns.GetHostAddressesAsync;
|
||||
return new SocketsHttpHandler
|
||||
{
|
||||
AllowAutoRedirect = false,
|
||||
PooledConnectionLifetime = TimeSpan.FromMinutes(5),
|
||||
ConnectCallback = async (context, cancellationToken) =>
|
||||
{
|
||||
var host = context.DnsEndPoint.Host;
|
||||
var addresses = IPAddress.TryParse(host, out var ip) ? [ip] : await resolver(host, cancellationToken).ConfigureAwait(false);
|
||||
if (addresses.Length == 0) throw new HttpRequestException($"{host} does not resolve");
|
||||
if (!allowPrivate() && addresses.Any(IsPrivate))
|
||||
throw new HttpRequestException($"{host} resolves to a private or loopback address");
|
||||
var socket = new Socket(SocketType.Stream, ProtocolType.Tcp) { NoDelay = true };
|
||||
try
|
||||
{
|
||||
await socket.ConnectAsync(addresses, context.DnsEndPoint.Port, cancellationToken).ConfigureAwait(false);
|
||||
return new NetworkStream(socket, ownsSocket: true);
|
||||
}
|
||||
catch
|
||||
{
|
||||
socket.Dispose();
|
||||
throw;
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@@ -31,3 +31,54 @@ public class PushUrlGuardTests
|
||||
public async Task Push_PrivateAllowed_Accepted() =>
|
||||
Assert.Null(await Guard(true).ProblemAsync("http://127.0.0.1:5000/x", default));
|
||||
}
|
||||
|
||||
public class PushUrlGuardAddressTests
|
||||
{
|
||||
[Theory]
|
||||
[InlineData("224.0.0.1", true)]
|
||||
[InlineData("239.255.255.250", true)]
|
||||
[InlineData("240.0.0.1", true)]
|
||||
[InlineData("255.255.255.255", true)]
|
||||
[InlineData("198.18.0.1", true)]
|
||||
[InlineData("198.19.255.254", true)]
|
||||
[InlineData("ff02::1", true)]
|
||||
[InlineData("::1", true)]
|
||||
[InlineData("fe80::1", true)]
|
||||
[InlineData("198.20.0.1", false)]
|
||||
[InlineData("93.184.216.34", false)]
|
||||
[InlineData("2606:2800:220:1::1", false)]
|
||||
public void IsPrivate_ClassifiesRanges(string address, bool expected) =>
|
||||
Assert.Equal(expected, PushUrlGuard.IsPrivate(IPAddress.Parse(address)));
|
||||
|
||||
[Fact]
|
||||
public async Task GuardedHandler_HostResolvingToLoopback_Refused()
|
||||
{
|
||||
using var http = new HttpClient(PushUrlGuard.CreateGuardedHandler(() => false,
|
||||
(_, _) => Task.FromResult(new[] { IPAddress.Loopback })));
|
||||
var ex = await Assert.ThrowsAsync<HttpRequestException>(() => http.GetAsync("http://hooks.example:1/x"));
|
||||
Assert.Contains("private", ex.ToString());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void GuardedHandler_DoesNotFollowRedirects() =>
|
||||
Assert.False(PushUrlGuard.CreateGuardedHandler(() => false).AllowAutoRedirect);
|
||||
|
||||
[Theory]
|
||||
[InlineData("tok\r\nX-Evil: 1", null, "token")]
|
||||
[InlineData(null, "Bearer x", "scheme")]
|
||||
[InlineData(null, "Bea\nrer", "scheme")]
|
||||
public void HeaderProblem_RejectsInjection(string? token, string? scheme, string expected)
|
||||
{
|
||||
var config = new global::A2A.PushNotificationConfig { Url = "https://hooks.example/x", Token = token };
|
||||
if (scheme is not null) config.Authentication = new global::A2A.AuthenticationInfo { Scheme = scheme, Credentials = "c" };
|
||||
Assert.Contains(expected, PushUrlGuard.HeaderProblem(config));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void HeaderProblem_PlainValues_Accepted() =>
|
||||
Assert.Null(PushUrlGuard.HeaderProblem(new global::A2A.PushNotificationConfig
|
||||
{
|
||||
Url = "https://hooks.example/x", Token = "abc-123",
|
||||
Authentication = new global::A2A.AuthenticationInfo { Scheme = "Bearer", Credentials = "secret" },
|
||||
}));
|
||||
}
|
||||
|
||||
132
tests/Volcanic.AgenticCore.A2A.Server.Tests/ServerUnitTests.cs
Normal file
132
tests/Volcanic.AgenticCore.A2A.Server.Tests/ServerUnitTests.cs
Normal file
@@ -0,0 +1,132 @@
|
||||
using A2A;
|
||||
using Microsoft.AspNetCore.Http;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Volcanic.AgenticCore.A2A.Server.Callers;
|
||||
using Volcanic.AgenticCore.A2A.Server.Hosting;
|
||||
using Volcanic.AgenticCore.A2A.Server.Push;
|
||||
using Volcanic.AgenticCore.A2A.Server.Tasks;
|
||||
using Volcanic.AgenticCore.Features.Extensions;
|
||||
using Volcanic.AgenticCore.Hosting;
|
||||
using Volcanic.AgenticCore.Testing;
|
||||
|
||||
namespace Volcanic.AgenticCore.A2A.Server.Tests;
|
||||
|
||||
/// <summary>The per-caller server's own checks, without HTTP.</summary>
|
||||
public sealed class ServerUnitTests : IAsyncDisposable
|
||||
{
|
||||
private readonly string _root = Path.Combine(Path.GetTempPath(), "a2a-server-unit", Guid.NewGuid().ToString("N"));
|
||||
private readonly FauxProvider _faux = new();
|
||||
private readonly ServiceProvider _services;
|
||||
private readonly A2ACaller _caller;
|
||||
|
||||
private sealed class NoCallers : IA2ACallerDirectory { public IReadOnlyList<A2ACaller> Find(string key) => []; }
|
||||
private sealed class NotLive : IA2ALiveRuns { public bool IsLive(string taskId) => false; }
|
||||
|
||||
public ServerUnitTests()
|
||||
{
|
||||
_caller = new A2ACaller("alpha", "isolated", Path.Combine(_root, "ws"), "p", "1");
|
||||
Directory.CreateDirectory(_caller.Workspace);
|
||||
var services = new ServiceCollection();
|
||||
services.AddLogging();
|
||||
services.AddAgenticCore(o =>
|
||||
{
|
||||
o.AgentDir = Path.Combine(_root, ".agent");
|
||||
o.DiscoverExtensionAssemblies = false;
|
||||
o.DefaultModel = FauxProvider.ModelKey;
|
||||
o.Extensions.Add(ExtensionRegistration.FromInstance(_faux.Extension));
|
||||
});
|
||||
services.AddSingleton<IA2ACallerDirectory, NoCallers>();
|
||||
services.AddA2AServer(o => o.AllowPrivatePushUrls = true);
|
||||
_services = services.BuildServiceProvider();
|
||||
}
|
||||
|
||||
private VolcanicA2AServer Server => _services.GetRequiredService<A2AServerCache>().For(_caller);
|
||||
|
||||
private Task SeedAsync(string taskId, TaskState state, string contextId = "c1") =>
|
||||
new WorkspaceTaskStore(_caller, new NotLive(), A2APushSender.Disabled, TimeProvider.System, NullLogger<WorkspaceTaskStore>.Instance)
|
||||
.SaveTaskAsync(taskId, new AgentTask
|
||||
{
|
||||
Id = taskId, ContextId = contextId,
|
||||
Status = new global::A2A.TaskStatus { State = state, Timestamp = DateTimeOffset.UtcNow },
|
||||
});
|
||||
|
||||
private static SendMessageRequest Send(string? taskId, string? contextId) => new()
|
||||
{
|
||||
Message = new Message
|
||||
{
|
||||
Role = Role.User, MessageId = Guid.NewGuid().ToString("N"), TaskId = taskId, ContextId = contextId,
|
||||
Parts = [Part.FromText("hi")],
|
||||
},
|
||||
};
|
||||
|
||||
[Fact]
|
||||
public void AddA2AServer_RegistersNoBareHttpClient() => Assert.Null(_services.GetService<HttpClient>());
|
||||
|
||||
[Fact]
|
||||
public async Task Continuation_OtherContext_InvalidParams()
|
||||
{
|
||||
await SeedAsync("t1", TaskState.InputRequired, "c1");
|
||||
var ex = await Assert.ThrowsAsync<A2AException>(() => Server.SendMessageAsync(Send("t1", "c2")));
|
||||
Assert.Equal(A2AErrorCode.InvalidParams, ex.ErrorCode);
|
||||
Assert.Contains("another context", ex.Message);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Continuation_TaskNotWaiting_Unsupported()
|
||||
{
|
||||
await SeedAsync("t2", TaskState.Working, "c1");
|
||||
var ex = await Assert.ThrowsAsync<A2AException>(() => Server.SendMessageAsync(Send("t2", null)));
|
||||
Assert.Equal(A2AErrorCode.UnsupportedOperation, ex.ErrorCode);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task PushConfig_GetNull_ReturnsNull() =>
|
||||
Assert.Null(await Server.GetTaskPushNotificationConfigAsync(null!));
|
||||
|
||||
[Fact]
|
||||
public async Task PushConfig_ListOrDelete_UnknownTask_NotFound()
|
||||
{
|
||||
var list = await Assert.ThrowsAsync<A2AException>(() =>
|
||||
Server.ListTaskPushNotificationConfigAsync(new ListTaskPushNotificationConfigRequest { TaskId = "nope" }));
|
||||
Assert.Equal(A2AErrorCode.TaskNotFound, list.ErrorCode);
|
||||
var delete = await Assert.ThrowsAsync<A2AException>(() =>
|
||||
Server.DeleteTaskPushNotificationConfigAsync(new DeleteTaskPushNotificationConfigRequest { TaskId = "nope", Id = "x" }));
|
||||
Assert.Equal(A2AErrorCode.TaskNotFound, delete.ErrorCode);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task PushConfig_CreateWithoutId_GetsGeneratedId_AndBadTokenRejected()
|
||||
{
|
||||
await SeedAsync("t3", TaskState.Completed);
|
||||
var created = await Server.CreateTaskPushNotificationConfigAsync(new CreateTaskPushNotificationConfigRequest
|
||||
{
|
||||
TaskId = "t3", Config = new PushNotificationConfig { Url = "https://hooks.example/x" },
|
||||
});
|
||||
Assert.False(string.IsNullOrEmpty(created.Id));
|
||||
var ex = await Assert.ThrowsAsync<A2AException>(() => Server.CreateTaskPushNotificationConfigAsync(new CreateTaskPushNotificationConfigRequest
|
||||
{
|
||||
TaskId = "t3", Config = new PushNotificationConfig { Url = "https://hooks.example/x", Token = "a\r\nb" },
|
||||
}));
|
||||
Assert.Equal(A2AErrorCode.InvalidParams, ex.ErrorCode);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Router_ExtendedCard_UsesCurrentRequestBaseUrl()
|
||||
{
|
||||
var context = new DefaultHttpContext();
|
||||
context.Request.Scheme = "http";
|
||||
context.Request.Host = new HostString("agent.test:8080");
|
||||
A2AAuth.SetCaller(context, _caller);
|
||||
var router = new A2ARequestRouter(new HttpContextAccessor { HttpContext = context },
|
||||
_services.GetRequiredService<A2AServerCache>(), _services.GetRequiredService<A2ACardProvider>());
|
||||
var card = await router.GetExtendedAgentCardAsync(new GetExtendedAgentCardRequest());
|
||||
Assert.Equal("http://agent.test:8080/a2a", card.SupportedInterfaces[0].Url);
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await _services.DisposeAsync();
|
||||
try { Directory.Delete(_root, true); } catch (Exception) { }
|
||||
}
|
||||
}
|
||||
@@ -48,7 +48,7 @@ public sealed class SessionPoolTests : IAsyncDisposable
|
||||
}
|
||||
|
||||
private A2AAgentHandler Handler() =>
|
||||
new(_caller, _pool, Options.Create(_options), SecretRedactor.Empty, new HttpClient(), TimeProvider.System,
|
||||
new(_caller, _pool, Options.Create(_options), SecretRedactor.Empty, new A2AFileDownloader(new HttpClient()), TimeProvider.System,
|
||||
NullLogger<A2AAgentHandler>.Instance);
|
||||
|
||||
private static RequestContext Context(string text, string taskId = "t1", string contextId = "c1") => new()
|
||||
|
||||
Reference in New Issue
Block a user