microsoft/semantic-kernel
1
0
Fork 0
mirror of https://github.com/microsoft/semantic-kernel.git synced 2026-10-12 02:19:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Python: .NET/Python: Share HTTP request validation utilities (#14546)

Browse Source 
### Motivation and Context

<!-- Thank you for your contribution to the semantic-kernel repo!
Please help reviewers and future users, providing the following
information:
  1. Why is this change required?
  2. What problem does it solve?
  3. What scenario does it contribute to?
  4. If it fixes an open issue, please link to the issue here.
-->

Improve consistency and maintainability of HTTP request validation in
the .NET and Python SDKs.

### Description

<!-- Describe your changes, the overall approach, the underlying design.
These notes will help understanding how your code works. Thanks! -->

- Extract existing request validation into shared internal utilities and
reuse them in the HTTP plugins.
- Update HTTP plugin configuration and documentation.
- Extend unit coverage for shared validation and plugin behavior.

### Contribution Checklist

<!-- Before submitting this PR, please make sure: -->

- [x] The code builds clean without any errors or warnings
- [ ] The PR follows the [SK Contribution
Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md)
and the [pre-submission formatting
script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts)
raises no violations
- [ ] All unit tests pass, and I have added new tests where possible
- [ ] I didn't break anyone 😄

---------

Co-authored-by: Copilot App <[email protected]>
This commit is contained in:
SergeyMenshykh
2026-10-06 10:08:03 +00:00
committed by GitHub
co-authored by Copilot App
parent d5234384a0
commit 71b2193130
11 changed files with 1049 additions and 491 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/src/Functions/Functions.OpenApi/ServerUrlValidator.cs
+6 -294
View File
@@ -3,9 +3,9 @@
using System;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.SemanticKernel.Http;
namespace Microsoft.SemanticKernel.Plugins.OpenApi;
@@ -71,7 +71,11 @@ internal static class ServerUrlValidator
return;
}
await EnsurePublicHostAsync(url, cancellationToken, dnsResolver).ConfigureAwait(false);
await PublicNetworkAddressValidator.ValidateAsync(
url, cancellationToken, dnsResolver,
"To allow this URL, add it to " +
$"{nameof(RestApiOperationServerUrlValidationOptions)}.{nameof(RestApiOperationServerUrlValidationOptions.AllowedBaseUrls)} " +
$"or set {nameof(RestApiOperationServerUrlValidationOptions)}.{nameof(RestApiOperationServerUrlValidationOptions.AllowPrivateNetworkAccess)} = true.").ConfigureAwait(false);
}
private static bool TryMatchAllowedBaseUrl(Uri url, IReadOnlyList<Uri>? allowedBaseUrls)
@@ -112,296 +116,4 @@ internal static class ServerUrlValidator
return false;
}
private static async Task EnsurePublicHostAsync(
Uri url,
CancellationToken cancellationToken,
Func<string, CancellationToken, Task<IPAddress[]>>? dnsResolver)
{
var host = url.DnsSafeHost;
// Case 1: literal IP in the URL.
if (url.HostNameType == UriHostNameType.IPv4 || url.HostNameType == UriHostNameType.IPv6)
{
if (!IPAddress.TryParse(host, out var ip))
{
// Should be unreachable — .NET's Uri already classified this as an IP address.
// Fail closed: block the request rather than silently skipping validation.
throw new InvalidOperationException(
$"The server URL '{url}' has a host identified as an IP address but could not be parsed. The request is blocked as a precaution.");
}
EnsurePublicAddress(url, ip);
return;
}
// Case 2: hostname - resolve and validate every returned address (defeats DNS rebinding).
IPAddress[] addresses;
try
{
if (dnsResolver is not null)
{
addresses = await dnsResolver(host, cancellationToken).ConfigureAwait(false);
}
else
{
#if NET
addresses = await Dns.GetHostAddressesAsync(host, cancellationToken).ConfigureAwait(false);
#else
addresses = await Dns.GetHostAddressesAsync(host).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
#endif
}
}
catch (SocketException)
{
// DNS resolution failed. Fail closed: block the request rather than allowing
// it to proceed unvalidated. A TOCTOU race could otherwise allow a private
// target to be reached if DNS fails here but succeeds when HttpClient resolves.
throw new InvalidOperationException(
$"The request URI '{url}' is not allowed: DNS resolution for host '{host}' failed. " +
"The request is blocked as a precaution to prevent potential access to private network addresses. " +
"To allow this URL, add it to " +
$"{nameof(RestApiOperationServerUrlValidationOptions)}.{nameof(RestApiOperationServerUrlValidationOptions.AllowedBaseUrls)} " +
$"or set {nameof(RestApiOperationServerUrlValidationOptions)}.{nameof(RestApiOperationServerUrlValidationOptions.AllowPrivateNetworkAccess)} = true.");
}
if (addresses is null || addresses.Length == 0)
{
// No addresses returned. Fail closed for the same TOCTOU reason.
throw new InvalidOperationException(
$"The request URI '{url}' is not allowed: DNS resolution for host '{host}' returned no addresses. " +
"The request is blocked as a precaution. To allow this URL, add it to " +
$"{nameof(RestApiOperationServerUrlValidationOptions)}.{nameof(RestApiOperationServerUrlValidationOptions.AllowedBaseUrls)} " +
$"or set {nameof(RestApiOperationServerUrlValidationOptions)}.{nameof(RestApiOperationServerUrlValidationOptions.AllowPrivateNetworkAccess)} = true.");
}
foreach (var address in addresses)
{
EnsurePublicAddress(url, address);
}
}
private static void EnsurePublicAddress(Uri url, IPAddress address)
{
if (TryCategorizeNonPublicAddress(address, out var category))
{
throw new InvalidOperationException(
$"The request URI '{url}' is not allowed: host resolves to a {category} address ({address}), " +
"which is blocked by default to prevent Server-Side Request Forgery (SSRF). " +
"To allow this URL, add it to " +
$"{nameof(RestApiOperationServerUrlValidationOptions)}.{nameof(RestApiOperationServerUrlValidationOptions.AllowedBaseUrls)} " +
$"or set {nameof(RestApiOperationServerUrlValidationOptions)}.{nameof(RestApiOperationServerUrlValidationOptions.AllowPrivateNetworkAccess)} = true.");
}
}
/// <summary>
/// Returns true and sets <paramref name="category"/> to a human-readable label when the
/// supplied address is in a non-public IP range that should be blocked by default.
/// </summary>
internal static bool TryCategorizeNonPublicAddress(IPAddress address, out string category)
{
// Normalize IPv4-mapped IPv6 (::ffff:a.b.c.d) to its IPv4 form so the v4 checks apply.
if (address.AddressFamily == AddressFamily.InterNetworkV6 && IsIPv4MappedToIPv6(address))
{
address = MapToIPv4(address);
}
if (address.AddressFamily == AddressFamily.InterNetwork)
{
return TryClassifyIPv4(address, out category);
}
if (address.AddressFamily == AddressFamily.InterNetworkV6)
{
return TryClassifyIPv6(address, out category);
}
// Not IPv4 or IPv6 - reject conservatively.
category = "non-IP";
return true;
}
private static bool TryClassifyIPv4(IPAddress address, out string category)
{
var bytes = address.GetAddressBytes();
var b0 = bytes[0];
var b1 = bytes[1];
var b2 = bytes[2];
// 0.0.0.0/8 - "this network", unspecified.
if (b0 == 0)
{
category = "unspecified";
return true;
}
// 10.0.0.0/8
if (b0 == 10)
{
category = "private (RFC1918)";
return true;
}
// 127.0.0.0/8 - loopback.
if (b0 == 127)
{
category = "loopback";
return true;
}
// 169.254.0.0/16 - link-local (includes cloud metadata 169.254.169.254).
if (b0 == 169 && b1 == 254)
{
category = "link-local";
return true;
}
// 172.16.0.0/12
if (b0 == 172 && b1 >= 16 && b1 <= 31)
{
category = "private (RFC1918)";
return true;
}
// 192.168.0.0/16
if (b0 == 192 && b1 == 168)
{
category = "private (RFC1918)";
return true;
}
// 100.64.0.0/10 - carrier-grade NAT.
if (b0 == 100 && b1 >= 64 && b1 <= 127)
{
category = "carrier-grade NAT";
return true;
}
// 198.18.0.0/15 - benchmarking.
if (b0 == 198 && (b1 == 18 || b1 == 19))
{
category = "benchmarking";
return true;
}
// 192.0.0.0/24 (IETF protocol assignments) and 192.0.2.0/24 (TEST-NET-1).
if (b0 == 192 && b1 == 0 && (b2 == 0 || b2 == 2))
{
category = "reserved";
return true;
}
// 198.51.100.0/24 (TEST-NET-2).
if (b0 == 198 && b1 == 51 && b2 == 100)
{
category = "reserved";
return true;
}
// 203.0.113.0/24 (TEST-NET-3).
if (b0 == 203 && b1 == 0 && b2 == 113)
{
category = "reserved";
return true;
}
// 224.0.0.0/4 - multicast.
if (b0 >= 224 && b0 <= 239)
{
category = "multicast";
return true;
}
// 240.0.0.0/4 - reserved (includes 255.255.255.255 broadcast).
if (b0 >= 240)
{
category = "reserved";
return true;
}
category = string.Empty;
return false;
}
private static bool TryClassifyIPv6(IPAddress address, out string category)
{
if (IPAddress.IsLoopback(address))
{
category = "loopback";
return true;
}
var bytes = address.GetAddressBytes();
// :: (unspecified)
if (address.Equals(IPAddress.IPv6None))
{
category = "unspecified";
return true;
}
// fe80::/10 - link-local.
if (bytes[0] == 0xfe && (bytes[1] & 0xC0) == 0x80)
{
category = "link-local";
return true;
}
// fc00::/7 - unique local (private).
if ((bytes[0] & 0xfe) == 0xfc)
{
category = "private (IPv6 ULA)";
return true;
}
// ff00::/8 - multicast.
if (bytes[0] == 0xff)
{
category = "multicast";
return true;
}
// 2001:db8::/32 - documentation.
if (bytes[0] == 0x20 && bytes[1] == 0x01 && bytes[2] == 0x0d && bytes[3] == 0xb8)
{
category = "reserved";
return true;
}
category = string.Empty;
return false;
}
private static bool IsIPv4MappedToIPv6(IPAddress address)
{
#if NET
return address.IsIPv4MappedToIPv6;
#else
var bytes = address.GetAddressBytes();
if (bytes.Length != 16)
{
return false;
}
for (int i = 0; i < 10; i++)
{
if (bytes[i] != 0)
{
return false;
}
}
return bytes[10] == 0xff && bytes[11] == 0xff;
#endif
}
private static IPAddress MapToIPv4(IPAddress address)
{
#if NET
return address.MapToIPv4();
#else
var bytes = address.GetAddressBytes();
var v4 = new byte[4] { bytes[12], bytes[13], bytes[14], bytes[15] };
return new IPAddress(v4);
#endif
}
}
dotnet/src/Functions/Functions.UnitTests/OpenApi/ServerUrlValidatorTests.cs
+75 -22
View File
@@ -45,24 +45,29 @@ public class ServerUrlValidatorTests
[InlineData("::1", "loopback")]
[InlineData("::", "unspecified")]
[InlineData("fe80::1", "link-local")]
[InlineData("febf:ffff:ffff:ffff:ffff:ffff:ffff:ffff", "link-local")]
[InlineData("fec0::", "site-local")]
[InlineData("fec0::1", "site-local")]
[InlineData("feff:ffff:ffff:ffff:ffff:ffff:ffff:ffff", "site-local")]
[InlineData("fc00::1", "private (IPv6 ULA)")]
[InlineData("fd00::1", "private (IPv6 ULA)")]
[InlineData("ff00::", "multicast")]
[InlineData("ff02::1", "multicast")]
[InlineData("2001:db8::1", "reserved")]
// IPv4-mapped IPv6 of a private address
[InlineData("::ffff:127.0.0.1", "loopback")]
[InlineData("::ffff:169.254.169.254", "link-local")]
public void ItShouldClassifyNonPublicAddresses(string address, string expectedCategory)
public async Task ItShouldRejectNonPublicAddressesAsync(string address, string expectedCategory)
{
// Arrange
var ip = IPAddress.Parse(address);
var url = new UriBuilder(Uri.UriSchemeHttps, address).Uri;
// Act
var blocked = ServerUrlValidator.TryCategorizeNonPublicAddress(ip, out var category);
var exception = await Assert.ThrowsAsync<InvalidOperationException>(() =>
ServerUrlValidator.ValidateAsync(url, options: null));
// Assert
Assert.True(blocked, $"Expected {address} to be classified as non-public.");
Assert.Equal(expectedCategory, category);
Assert.Contains($"host resolves to a {expectedCategory} address", exception.Message, StringComparison.Ordinal);
}
[Theory]
@@ -78,16 +83,13 @@ public class ServerUrlValidatorTests
[InlineData("100.128.0.1")] // just outside CGNAT
// Public IPv6
[InlineData("2606:4700:4700::1111")] // Cloudflare DNS
public void ItShouldNotBlockPublicAddresses(string address)
public async Task ItShouldNotBlockPublicAddressesAsync(string address)
{
// Arrange
var ip = IPAddress.Parse(address);
var url = new UriBuilder(Uri.UriSchemeHttps, address).Uri;
// Act
var blocked = ServerUrlValidator.TryCategorizeNonPublicAddress(ip, out _);
// Assert
Assert.False(blocked, $"Expected {address} to be treated as public.");
// Act & Assert
await ServerUrlValidator.ValidateAsync(url, options: null);
}
[Fact]
@@ -135,6 +137,20 @@ public class ServerUrlValidatorTests
await ServerUrlValidator.ValidateAsync(url, options);
}
[Fact]
public async Task ItShouldPreserveExplicitHostnameAllowlistDnsBypassAsync()
{
var url = new Uri("http://trusted.example.com/v1/orders");
var options = new RestApiOperationServerUrlValidationOptions
{
AllowedBaseUrls = [new Uri("http://trusted.example.com/v1")]
};
Task<IPAddress[]> UnexpectedResolver(string host, CancellationToken token) =>
throw new InvalidOperationException("An explicitly trusted URL should not be resolved.");
await ServerUrlValidator.ValidateAsync(url, options, dnsResolver: UnexpectedResolver);
}
[Fact]
public async Task ItShouldAllowPublicHttpsHostByDefaultAsync()
{
@@ -143,27 +159,30 @@ public class ServerUrlValidatorTests
await ServerUrlValidator.ValidateAsync(url, options: null);
}
[Fact]
public async Task ItShouldBypassPrivateGateWhenAllowPrivateNetworkAccessTrueAsync()
[Theory]
[InlineData("https://10.0.0.5/")]
[InlineData("https://[fec0::1]/")]
public async Task ItShouldBypassPrivateGateWhenAllowPrivateNetworkAccessTrueAsync(string url)
{
var url = new Uri("https://10.0.0.5/");
var options = new RestApiOperationServerUrlValidationOptions { AllowPrivateNetworkAccess = true };
await ServerUrlValidator.ValidateAsync(url, options);
await ServerUrlValidator.ValidateAsync(new Uri(url), options);
}
[Fact]
public async Task ItShouldBlockHostnameResolvingToPrivateIpAsync()
[Theory]
[InlineData("169.254.169.254", "link-local")]
[InlineData("fec0::", "site-local")]
[InlineData("fec0::1", "site-local")]
[InlineData("feff:ffff:ffff:ffff:ffff:ffff:ffff:ffff", "site-local")]
public async Task ItShouldBlockHostnameResolvingToPrivateIpAsync(string address, string expectedCategory)
{
// Simulates an attacker-controlled hostname (e.g., evil.com) resolving to the
// cloud metadata address — the most realistic SSRF vector.
var url = new Uri("https://evil.example.com/latest/meta-data/");
Task<IPAddress[]> FakeResolver(string _, CancellationToken _1) =>
Task.FromResult(new[] { IPAddress.Parse("169.254.169.254") });
Task.FromResult(new[] { IPAddress.Parse(address) });
var ex = await Assert.ThrowsAsync<InvalidOperationException>(() =>
ServerUrlValidator.ValidateAsync(url, options: null, dnsResolver: FakeResolver));
Assert.Contains("link-local", ex.Message, StringComparison.OrdinalIgnoreCase);
Assert.Contains(expectedCategory, ex.Message, StringComparison.OrdinalIgnoreCase);
}
[Fact]
@@ -229,4 +248,38 @@ public class ServerUrlValidatorTests
ServerUrlValidator.ValidateAsync(url, options: null, dnsResolver: FakeResolver));
Assert.Contains("no addresses", ex.Message, StringComparison.OrdinalIgnoreCase);
}
[Fact]
public async Task ItShouldPropagateDnsCancellationAsync()
{
using var source = new CancellationTokenSource();
source.Cancel();
var url = new Uri("https://api.example.com/");
Task<IPAddress[]> ResolveAsync(string host, CancellationToken token)
{
Assert.Equal(source.Token, token);
return Task.FromCanceled<IPAddress[]>(token);
}
await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
ServerUrlValidator.ValidateAsync(url, options: null, cancellationToken: source.Token, dnsResolver: ResolveAsync));
}
[Theory]
[InlineData("https://fa\u00df.test/", "xn--fa-hia.test")]
[InlineData("https://b\u00fccher.test/", "xn--bcher-kva.test")]
[InlineData("https://\uFF25\uFF38\uFF21\uFF2D\uFF30\uFF2C\uFF25.com/", "example.com")]
public async Task ItShouldResolveTheSameAsciiHostAsHttpClientAsync(string url, string expectedHost)
{
string? resolvedHost = null;
Task<IPAddress[]> ResolveAsync(string host, CancellationToken _)
{
resolvedHost = host;
return Task.FromResult(new[] { IPAddress.Parse("93.184.216.34") });
}
await ServerUrlValidator.ValidateAsync(new Uri(url), options: null, dnsResolver: ResolveAsync);
Assert.Equal(expectedHost, resolvedHost);
}
}
dotnet/src/InternalUtilities/src/Http/PublicNetworkAddressValidator.cs
+318
View File
@@ -0,0 +1,318 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Diagnostics.CodeAnalysis;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
namespace Microsoft.SemanticKernel.Http;
/// <summary>
/// Resolves and rejects non-public request destinations before an HTTP request is sent.
/// This check does not pin the validated addresses to the subsequent HTTP connection.
/// </summary>
[ExcludeFromCodeCoverage]
internal static class PublicNetworkAddressValidator
{
/// <summary>
/// Validates all addresses of a request host, failing closed on DNS errors.
/// </summary>
public static async Task ValidateAsync(
Uri url,
CancellationToken cancellationToken = default,
Func<string, CancellationToken, Task<IPAddress[]>>? dnsResolver = null,
string configurationHint = "")
{
var host = url.DnsSafeHost;
// Case 1: literal IP in the URL.
if (url.HostNameType == UriHostNameType.IPv4 || url.HostNameType == UriHostNameType.IPv6)
{
if (!IPAddress.TryParse(host, out var ip))
{
// Should be unreachable — .NET's Uri already classified this as an IP address.
// Fail closed: block the request rather than silently skipping validation.
throw new InvalidOperationException(
$"The server URL '{url}' has a host identified as an IP address but could not be parsed. The request is blocked as a precaution.");
}
EnsurePublicAddress(url, ip, configurationHint);
return;
}
// Resolve the ASCII host used by HttpClient. DnsSafeHost can contain Unicode IDN labels.
host = url.IdnHost;
// Reject a mixed public/private DNS response as well as an entirely private one.
IPAddress[] addresses;
try
{
if (dnsResolver is not null)
{
addresses = await dnsResolver(host, cancellationToken).ConfigureAwait(false);
}
else
{
#if NET
addresses = await Dns.GetHostAddressesAsync(host, cancellationToken).ConfigureAwait(false);
#else
addresses = await Dns.GetHostAddressesAsync(host).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
#endif
}
}
catch (SocketException)
{
// DNS resolution failed. Fail closed: block the request rather than allowing
// it to proceed unvalidated. A TOCTOU race could otherwise allow a private
// target to be reached if DNS fails here but succeeds when HttpClient resolves.
throw new InvalidOperationException(
$"The request URI '{url}' is not allowed: DNS resolution for host '{host}' failed. " +
"The request is blocked as a precaution to prevent potential access to private network addresses. " +
configurationHint);
}
if (addresses is null || addresses.Length == 0)
{
// No addresses returned. Fail closed for the same TOCTOU reason.
throw new InvalidOperationException(
$"The request URI '{url}' is not allowed: DNS resolution for host '{host}' returned no addresses. " +
"The request is blocked as a precaution. " + configurationHint);
}
foreach (var address in addresses)
{
EnsurePublicAddress(url, address, configurationHint);
}
}
/// <summary>
/// Returns true and sets <paramref name="category"/> to a human-readable label when the
/// supplied address is in a non-public IP range that should be blocked by default.
/// </summary>
internal static bool TryCategorizeNonPublicAddress(IPAddress address, out string category)
{
// Normalize IPv4-mapped IPv6 (::ffff:a.b.c.d) to its IPv4 form so the v4 checks apply.
if (address.AddressFamily == AddressFamily.InterNetworkV6 && IsIPv4MappedToIPv6(address))
{
address = MapToIPv4(address);
}
if (address.AddressFamily == AddressFamily.InterNetwork)
{
return TryClassifyIPv4(address, out category);
}
if (address.AddressFamily == AddressFamily.InterNetworkV6)
{
return TryClassifyIPv6(address, out category);
}
// Not IPv4 or IPv6 - reject conservatively.
category = "non-IP";
return true;
}
private static void EnsurePublicAddress(Uri url, IPAddress address, string configurationHint)
{
if (TryCategorizeNonPublicAddress(address, out var category))
{
throw new InvalidOperationException(
$"The request URI '{url}' is not allowed: host resolves to a {category} address ({address}), " +
"which is blocked by default to prevent Server-Side Request Forgery (SSRF). " +
configurationHint);
}
}
private static bool TryClassifyIPv4(IPAddress address, out string category)
{
var bytes = address.GetAddressBytes();
var b0 = bytes[0];
var b1 = bytes[1];
var b2 = bytes[2];
// 0.0.0.0/8 - "this network", unspecified.
if (b0 == 0)
{
category = "unspecified";
return true;
}
// 10.0.0.0/8
if (b0 == 10)
{
category = "private (RFC1918)";
return true;
}
// 127.0.0.0/8 - loopback.
if (b0 == 127)
{
category = "loopback";
return true;
}
// 169.254.0.0/16 - link-local (includes cloud metadata 169.254.169.254).
if (b0 == 169 && b1 == 254)
{
category = "link-local";
return true;
}
// 172.16.0.0/12
if (b0 == 172 && b1 >= 16 && b1 <= 31)
{
category = "private (RFC1918)";
return true;
}
// 192.168.0.0/16
if (b0 == 192 && b1 == 168)
{
category = "private (RFC1918)";
return true;
}
// 100.64.0.0/10 - carrier-grade NAT.
if (b0 == 100 && b1 >= 64 && b1 <= 127)
{
category = "carrier-grade NAT";
return true;
}
// 198.18.0.0/15 - benchmarking.
if (b0 == 198 && (b1 == 18 || b1 == 19))
{
category = "benchmarking";
return true;
}
// 192.0.0.0/24 (IETF protocol assignments) and 192.0.2.0/24 (TEST-NET-1).
if (b0 == 192 && b1 == 0 && (b2 == 0 || b2 == 2))
{
category = "reserved";
return true;
}
// 198.51.100.0/24 (TEST-NET-2).
if (b0 == 198 && b1 == 51 && b2 == 100)
{
category = "reserved";
return true;
}
// 203.0.113.0/24 (TEST-NET-3).
if (b0 == 203 && b1 == 0 && b2 == 113)
{
category = "reserved";
return true;
}
// 224.0.0.0/4 - multicast.
if (b0 >= 224 && b0 <= 239)
{
category = "multicast";
return true;
}
// 240.0.0.0/4 - reserved (includes 255.255.255.255 broadcast).
if (b0 >= 240)
{
category = "reserved";
return true;
}
category = string.Empty;
return false;
}
private static bool TryClassifyIPv6(IPAddress address, out string category)
{
if (IPAddress.IsLoopback(address))
{
category = "loopback";
return true;
}
var bytes = address.GetAddressBytes();
// :: (unspecified)
if (address.Equals(IPAddress.IPv6None))
{
category = "unspecified";
return true;
}
// fe80::/10 - link-local.
if (bytes[0] == 0xfe && (bytes[1] & 0xC0) == 0x80)
{
category = "link-local";
return true;
}
// fec0::/10 - deprecated site-local.
if (bytes[0] == 0xfe && (bytes[1] & 0xC0) == 0xC0)
{
category = "site-local";
return true;
}
// fc00::/7 - unique local (private).
if ((bytes[0] & 0xfe) == 0xfc)
{
category = "private (IPv6 ULA)";
return true;
}
// ff00::/8 - multicast.
if (bytes[0] == 0xff)
{
category = "multicast";
return true;
}
// 2001:db8::/32 - documentation.
if (bytes[0] == 0x20 && bytes[1] == 0x01 && bytes[2] == 0x0d && bytes[3] == 0xb8)
{
category = "reserved";
return true;
}
category = string.Empty;
return false;
}
private static bool IsIPv4MappedToIPv6(IPAddress address)
{
#if NET
return address.IsIPv4MappedToIPv6;
#else
var bytes = address.GetAddressBytes();
if (bytes.Length != 16)
{
return false;
}
for (int i = 0; i < 10; i++)
{
if (bytes[i] != 0)
{
return false;
}
}
return bytes[10] == 0xff && bytes[11] == 0xff;
#endif
}
private static IPAddress MapToIPv4(IPAddress address)
{
#if NET
return address.MapToIPv4();
#else
var bytes = address.GetAddressBytes();
var v4 = new byte[4] { bytes[12], bytes[13], bytes[14], bytes[15] };
return new IPAddress(v4);
#endif
}
}
dotnet/src/Plugins/Plugins.Core/HttpPlugin.cs
+25
View File
@@ -26,6 +26,14 @@ namespace Microsoft.SemanticKernel.Plugins.Core;
/// <para>
/// The default HTTP client does not follow redirects to prevent bypassing the allow-list.
/// </para>
/// <para>
/// Allowed hosts are resolved before each request and non-public addresses are rejected unless
/// <see cref="AllowPrivateNetworkAccess"/> is explicitly enabled. DNS failures are rejected.
/// </para>
/// <para>
/// This pre-request check does not pin the connection's IP address and does not
/// protect against DNS changing between validation and connection.
/// </para>
/// </remarks>
[System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1054:URI-like parameters should not be strings",
Justification = "Semantic Kernel operates on strings")]
@@ -66,6 +74,16 @@ public sealed class HttpPlugin
set => this._allowedDomains = value is null ? null : new HashSet<string>(value, StringComparer.OrdinalIgnoreCase);
}
/// <summary>
/// Gets or sets whether allowed domains may resolve to private, loopback, link-local,
/// or other non-public addresses. Defaults to <see langword="false"/>.
/// </summary>
/// <remarks>
/// Enable only for trusted internal endpoints. Domain restrictions still apply and
/// the default HTTP client still rejects redirects.
/// </remarks>
public bool AllowPrivateNetworkAccess { get; set; }
/// <summary>
/// Sends an HTTP GET request to the specified URI and returns the response body as a string.
/// </summary>
@@ -147,6 +165,13 @@ public sealed class HttpPlugin
throw new InvalidOperationException("Sending requests to the provided location is not allowed.");
}
if (!this.AllowPrivateNetworkAccess)
{
await PublicNetworkAddressValidator.ValidateAsync(
uri, cancellationToken,
configurationHint: $"To allow trusted internal endpoints, set {nameof(this.AllowPrivateNetworkAccess)} = true.").ConfigureAwait(false);
}
using var request = new HttpRequestMessage(method, uri) { Content = requestContent };
request.Headers.Add("User-Agent", HttpHeaderConstant.Values.UserAgent);
request.Headers.Add(HttpHeaderConstant.Names.SemanticKernelVersion, HttpHeaderConstant.Values.GetAssemblyVersion(typeof(HttpPlugin)));
dotnet/src/Plugins/Plugins.UnitTests/Core/HttpPluginTests.cs
+136 -6
View File
@@ -16,7 +16,7 @@ namespace SemanticKernel.Plugins.UnitTests.Core;
public sealed class HttpPluginTests : IDisposable
{
private readonly string _content = "hello world";
private readonly string _uriString = "http://www.example.com";
private readonly string _uriString = "http://1.1.1.1";
private readonly HttpResponseMessage _response = new()
{
@@ -44,7 +44,7 @@ public sealed class HttpPluginTests : IDisposable
// Arrange
var mockHandler = this.CreateMock();
using var client = new HttpClient(mockHandler.Object);
var plugin = new HttpPlugin(client) { AllowedDomains = ["www.example.com"] };
var plugin = new HttpPlugin(client) { AllowedDomains = ["1.1.1.1"] };
// Act
var result = await plugin.GetAsync(this._uriString);
@@ -60,7 +60,7 @@ public sealed class HttpPluginTests : IDisposable
// Arrange
var mockHandler = this.CreateMock();
using var client = new HttpClient(mockHandler.Object);
var plugin = new HttpPlugin(client) { AllowedDomains = ["www.example.com"] };
var plugin = new HttpPlugin(client) { AllowedDomains = ["1.1.1.1"] };
// Act
var result = await plugin.PostAsync(this._uriString, this._content);
@@ -76,7 +76,7 @@ public sealed class HttpPluginTests : IDisposable
// Arrange
var mockHandler = this.CreateMock();
using var client = new HttpClient(mockHandler.Object);
var plugin = new HttpPlugin(client) { AllowedDomains = ["www.example.com"] };
var plugin = new HttpPlugin(client) { AllowedDomains = ["1.1.1.1"] };
// Act
var result = await plugin.PutAsync(this._uriString, this._content);
@@ -92,7 +92,7 @@ public sealed class HttpPluginTests : IDisposable
// Arrange
var mockHandler = this.CreateMock();
using var client = new HttpClient(mockHandler.Object);
var plugin = new HttpPlugin(client) { AllowedDomains = ["www.example.com"] };
var plugin = new HttpPlugin(client) { AllowedDomains = ["1.1.1.1"] };
// Act
var result = await plugin.DeleteAsync(this._uriString);
@@ -141,7 +141,8 @@ public sealed class HttpPluginTests : IDisposable
var plugin = new HttpPlugin()
{
AllowedDomains = [server.BaseUri.Host]
AllowedDomains = [server.BaseUri.Host],
AllowPrivateNetworkAccess = true
};
// Act & Assert - the plugin should throw because 302 is a non-success status
@@ -149,6 +150,135 @@ public sealed class HttpPluginTests : IDisposable
Assert.False(server.RedirectTargetContacted, "The redirect target should not have been contacted.");
}
[Theory]
[InlineData("GET")]
[InlineData("POST")]
[InlineData("PUT")]
[InlineData("DELETE")]
public async Task ItBlocksNonPublicAddressesBeforeSendingAsync(string method)
{
string[] addresses =
[
"169.254.169.254", "127.0.0.1", "10.0.0.1", "172.16.0.1", "192.168.0.1",
"100.64.0.1", "0.0.0.0", "224.0.0.1", "198.18.0.1", "192.0.2.1",
"::1", "fe80::1", "fec0::", "fec0::1", "feff:ffff:ffff:ffff:ffff:ffff:ffff:ffff",
"fc00::1", "ff02::1", "::ffff:169.254.169.254"
];
foreach (var address in addresses)
{
var mockHandler = this.CreateMock();
using var client = new HttpClient(mockHandler.Object);
var host = address.Contains(':') ? $"[{address}]" : address;
var uri = new Uri($"http://{host}/");
var plugin = new HttpPlugin(client) { AllowedDomains = [uri.Host] };
var exception = await Assert.ThrowsAsync<InvalidOperationException>(
() => SendAsync(plugin, method, uri.AbsoluteUri));
Assert.Contains("host resolves to", exception.Message, StringComparison.Ordinal);
Assert.Contains("AllowPrivateNetworkAccess", exception.Message, StringComparison.Ordinal);
VerifyNoRequest(mockHandler);
}
}
[Theory]
[InlineData("1.1.1.1")]
[InlineData("[2606:4700:4700::1111]")]
public async Task ItAllowsPublicLiteralAddressesAsync(string address)
{
var mockHandler = this.CreateMock();
using var client = new HttpClient(mockHandler.Object);
var uri = new Uri($"http://{address}/");
var plugin = new HttpPlugin(client) { AllowedDomains = [uri.Host] };
Assert.Equal(this._content, await plugin.GetAsync(uri.AbsoluteUri));
}
[Theory]
[InlineData("GET", "127.0.0.1")]
[InlineData("POST", "127.0.0.1")]
[InlineData("PUT", "127.0.0.1")]
[InlineData("DELETE", "127.0.0.1")]
[InlineData("GET", "[fec0::1]")]
[InlineData("POST", "[fec0::1]")]
[InlineData("PUT", "[fec0::1]")]
[InlineData("DELETE", "[fec0::1]")]
[InlineData("GET", "unresolved.example.invalid")]
[InlineData("POST", "unresolved.example.invalid")]
[InlineData("PUT", "unresolved.example.invalid")]
[InlineData("DELETE", "unresolved.example.invalid")]
public async Task ItAllowsExplicitPrivateNetworkAccessAsync(string method, string host)
{
var mockHandler = this.CreateMock();
using var client = new HttpClient(mockHandler.Object);
var plugin = new HttpPlugin(client)
{
AllowedDomains = [host],
AllowPrivateNetworkAccess = true
};
Assert.Equal(this._content, await SendAsync(plugin, method, $"http://{host}/"));
}
[Fact]
public async Task ItStillRestrictsDomainsWithPrivateNetworkAccessEnabledAsync()
{
var mockHandler = this.CreateMock();
using var client = new HttpClient(mockHandler.Object);
var plugin = new HttpPlugin(client)
{
AllowedDomains = ["www.example.com"],
AllowPrivateNetworkAccess = true
};
await Assert.ThrowsAsync<InvalidOperationException>(() => plugin.GetAsync("http://127.0.0.1/"));
VerifyNoRequest(mockHandler);
}
[Fact]
public async Task ItRejectsDisallowedDomainBeforeSendingAsync()
{
var mockHandler = this.CreateMock();
using var client = new HttpClient(mockHandler.Object);
var plugin = new HttpPlugin(client) { AllowedDomains = ["www.example.com"] };
var exception = await Assert.ThrowsAsync<InvalidOperationException>(() => plugin.GetAsync("http://notallowed.example.invalid/"));
Assert.Equal("Sending requests to the provided location is not allowed.", exception.Message);
VerifyNoRequest(mockHandler);
}
[Theory]
[InlineData("GET")]
[InlineData("POST")]
[InlineData("PUT")]
[InlineData("DELETE")]
public async Task ItBlocksAllowedLocalhostUsingDefaultDnsResolverAsync(string method)
{
var mockHandler = this.CreateMock();
using var client = new HttpClient(mockHandler.Object);
var plugin = new HttpPlugin(client) { AllowedDomains = ["localhost"] };
var exception = await Assert.ThrowsAsync<InvalidOperationException>(() => SendAsync(plugin, method, "http://localhost/"));
Assert.Contains("loopback", exception.Message, StringComparison.Ordinal);
VerifyNoRequest(mockHandler);
}
private static Task<string> SendAsync(HttpPlugin plugin, string method, string uri) =>
method switch
{
"GET" => plugin.GetAsync(uri),
"POST" => plugin.PostAsync(uri, "body"),
"PUT" => plugin.PutAsync(uri, "body"),
"DELETE" => plugin.DeleteAsync(uri),
_ => throw new ArgumentOutOfRangeException(nameof(method))
};
private static void VerifyNoRequest(Mock<HttpMessageHandler> mockHandler) =>
mockHandler.Protected().Verify(
"SendAsync", Times.Never(), ItExpr.IsAny<HttpRequestMessage>(), ItExpr.IsAny<CancellationToken>());
private Mock<HttpMessageHandler> CreateMock()
{
var mockHandler = new Mock<HttpMessageHandler>();
python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py
+9 -139
View File
@@ -1,9 +1,6 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import ipaddress
import socket
from collections.abc import Awaitable, Callable, Sequence
from typing import Any
from urllib.parse import ParseResult, urlparse
@@ -11,8 +8,7 @@ from pydantic import Field
from semantic_kernel.exceptions.function_exceptions import FunctionExecutionException
from semantic_kernel.kernel_pydantic import KernelBaseModel
DnsResolver = Callable[[str], Awaitable[Sequence[str | ipaddress.IPv4Address | ipaddress.IPv6Address]]]
from semantic_kernel.utils.public_network_address_validator import DnsResolver, ensure_public_host
DEFAULT_ALLOWED_SCHEME = "https"
@@ -70,22 +66,14 @@ async def validate_server_url(
if options.allow_private_network_access:
return []
return await _ensure_public_host(parsed_url, dns_resolver)
def try_categorize_non_public_address(
address: str | ipaddress.IPv4Address | ipaddress.IPv6Address,
) -> tuple[bool, str]:
"""Return whether an IP address is non-public and the category when blocked."""
ip_address = ipaddress.ip_address(address)
if isinstance(ip_address, ipaddress.IPv6Address) and ip_address.ipv4_mapped:
ip_address = ip_address.ipv4_mapped
if isinstance(ip_address, ipaddress.IPv4Address):
return _try_classify_ipv4(ip_address)
return _try_classify_ipv6(ip_address)
return await ensure_public_host(
parsed_url,
dns_resolver,
configuration_hint=(
"To allow this URL, add it to server_url_validation_allowed_base_urls or set "
"allow_private_network_access=True."
),
)
def _parse_absolute_url(url: str, option_name: str = "url") -> ParseResult:
@@ -134,121 +122,3 @@ def _matches_path_prefix(url_path: str, base_path: str) -> bool:
base_path_with_slash = base_path if base_path.endswith("/") else f"{base_path}/"
return url_path.lower().startswith(base_path_with_slash.lower())
async def _ensure_public_host(
parsed_url: ParseResult, dns_resolver: DnsResolver | None
) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]:
host = parsed_url.hostname
if host is None:
raise FunctionExecutionException(f"The request URI '{parsed_url.geturl()}' does not contain a valid host.")
try:
ip_address = ipaddress.ip_address(host)
except ValueError:
addresses = await _resolve_host(host, dns_resolver)
else:
_ensure_public_address(parsed_url.geturl(), ip_address)
# A literal IP address cannot be rebound between validation and connection.
return []
if not addresses:
raise FunctionExecutionException(
f"The request URI '{parsed_url.geturl()}' is not allowed: DNS resolution for host "
f"'{host}' returned no addresses. The request is blocked as a precaution."
)
for address in addresses:
_ensure_public_address(parsed_url.geturl(), address)
return addresses
async def _resolve_host(
host: str,
dns_resolver: DnsResolver | None,
) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]:
try:
if dns_resolver:
resolved_addresses = await dns_resolver(host)
return [ipaddress.ip_address(address) for address in resolved_addresses]
loop = asyncio.get_running_loop()
addr_info = await loop.getaddrinfo(host, None, type=socket.SOCK_STREAM)
except (OSError, ValueError) as exc:
raise FunctionExecutionException(
f"The request URI host '{host}' is not allowed: DNS resolution failed. "
"The request is blocked as a precaution to prevent potential access to private network addresses."
) from exc
addresses: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = []
seen_addresses: set[str] = set()
for family, _, _, _, sockaddr in addr_info:
if family not in (socket.AF_INET, socket.AF_INET6):
continue
address = ipaddress.ip_address(sockaddr[0])
address_string = str(address)
if address_string not in seen_addresses:
addresses.append(address)
seen_addresses.add(address_string)
return addresses
def _ensure_public_address(url: str, address: ipaddress.IPv4Address | ipaddress.IPv6Address) -> None:
blocked, category = try_categorize_non_public_address(address)
if blocked:
raise FunctionExecutionException(
f"The request URI '{url}' is not allowed: host resolves to a {category} address ({address}), "
"which is blocked by default to prevent Server-Side Request Forgery (SSRF). "
"To allow this URL, add it to server_url_validation_allowed_base_urls or set "
"allow_private_network_access=True."
)
def _try_classify_ipv4(address: ipaddress.IPv4Address) -> tuple[bool, str]:
b0, b1, b2, _ = address.packed
if b0 == 0:
return True, "unspecified"
if b0 == 10:
return True, "private (RFC1918)"
if b0 == 127:
return True, "loopback"
if b0 == 169 and b1 == 254:
return True, "link-local"
if b0 == 172 and 16 <= b1 <= 31:
return True, "private (RFC1918)"
if b0 == 192 and b1 == 168:
return True, "private (RFC1918)"
if b0 == 100 and 64 <= b1 <= 127:
return True, "carrier-grade NAT"
if b0 == 198 and b1 in (18, 19):
return True, "benchmarking"
if b0 == 192 and b1 == 0 and b2 in (0, 2):
return True, "reserved"
if b0 == 198 and b1 == 51 and b2 == 100:
return True, "reserved"
if b0 == 203 and b1 == 0 and b2 == 113:
return True, "reserved"
if 224 <= b0 <= 239:
return True, "multicast"
if b0 >= 240:
return True, "reserved"
return False, ""
def _try_classify_ipv6(address: ipaddress.IPv6Address) -> tuple[bool, str]:
if address.is_loopback:
return True, "loopback"
if address.is_unspecified:
return True, "unspecified"
if address.is_link_local:
return True, "link-local"
if address in ipaddress.ip_network("fc00::/7"):
return True, "private (IPv6 ULA)"
if address.is_multicast:
return True, "multicast"
if address in ipaddress.ip_network("2001:db8::/32"):
return True, "reserved"
return False, ""
python/semantic_kernel/core_plugins/http_plugin.py
+41 -13
View File
@@ -6,10 +6,12 @@ from typing import Annotated, Any, ClassVar
from urllib.parse import urlparse
import aiohttp
from yarl import URL
from semantic_kernel.exceptions import FunctionExecutionException
from semantic_kernel.functions.kernel_function_decorator import kernel_function
from semantic_kernel.kernel_pydantic import KernelBaseModel
from semantic_kernel.utils.public_network_address_validator import ensure_public_host
logger = logging.getLogger(__name__)
@@ -42,13 +44,26 @@ class HttpPlugin(KernelBaseModel):
- Only standard ports (80, 443) are permitted by default. Set ``allowed_ports``
to permit additional ports. Port validation is skipped when
``allow_all_domains`` is True.
- Allowed hosts are normalized using aiohttp's IDNA rules before DNS validation,
and the same normalized URL is used for the request. Non-public addresses and
DNS failures are blocked unless ``allow_private_network_access`` is True.
``allow_all_domains=True`` also bypasses this check.
- This pre-request check does not pin the connection's IP address and does not
protect against DNS changing between validation and connection.
"""
allowed_domains: set[str] | None = None
"""Set of allowed domains to send requests to."""
allow_all_domains: bool = False
"""When True, requests to any domain are allowed. Must be explicitly set."""
"""Allow any domain, port, and network address, with redirects. Must be explicitly set."""
allow_private_network_access: bool = False
"""Permit non-public addresses for allowed domains. Enable only for trusted internal endpoints.
Domain and port restrictions still apply, and redirects remain disabled unless
``allow_all_domains`` is True.
"""
allowed_ports: set[int] | None = None
"""Set of ports permitted for outbound requests. Defaults to ``{80, 443}`` when not set.
@@ -126,20 +141,24 @@ class HttpPlugin(KernelBaseModel):
# Default: deny all
return False
def _validate_url(self, url: str) -> None:
"""Validate the URL before sending a request.
async def _validate_url(self, url: str) -> URL:
"""Validate and normalize the URL before sending a request.
Always checks that the URL is non-empty, uses an allowed scheme, and has a
syntactically valid port. When ``allow_all_domains`` is False, additionally
enforces the port and domain allow-lists.
enforces the port and domain allow-lists. Non-public addresses and DNS failures
are rejected unless private-network access or unrestricted access is enabled.
Args:
url: The URL to validate.
Returns:
The normalized URL to use for the request.
Raises:
FunctionExecutionException: If the URL is empty, uses a disallowed scheme,
has a malformed port, or (unless ``allow_all_domains`` is True) targets
a port or domain that is not allowed.
a port or domain that is not allowed, or resolves to a prohibited address.
"""
if not url:
raise FunctionExecutionException("url cannot be `None` or empty")
@@ -147,6 +166,15 @@ class HttpPlugin(KernelBaseModel):
if not self._is_uri_allowed(url):
raise FunctionExecutionException("Sending requests to the provided location is not allowed.")
request_url = URL(url)
if not self.allow_all_domains and not self.allow_private_network_access:
await ensure_public_host(
urlparse(str(request_url)),
configuration_hint="To allow trusted internal endpoints, set allow_private_network_access=True.",
)
return request_url
@kernel_function(description="Makes a GET request to a url", name="getAsync")
async def get(self, url: Annotated[str, "The URL to send the request to."]) -> str:
"""Sends an HTTP GET request to the specified URI and returns the response body as a string.
@@ -157,11 +185,11 @@ class HttpPlugin(KernelBaseModel):
Returns:
The response body as a string.
"""
self._validate_url(url)
request_url = await self._validate_url(url)
async with (
aiohttp.ClientSession() as session,
session.get(url, raise_for_status=True, allow_redirects=self._allow_redirects) as response,
session.get(request_url, raise_for_status=True, allow_redirects=self._allow_redirects) as response,
):
return await response.text()
@@ -179,14 +207,14 @@ class HttpPlugin(KernelBaseModel):
returns:
The response body as a string.
"""
self._validate_url(url)
request_url = await self._validate_url(url)
headers = {"Content-Type": "application/json"}
data = json.dumps(body) if body is not None else None
async with (
aiohttp.ClientSession() as session,
session.post(
url, headers=headers, data=data, raise_for_status=True, allow_redirects=self._allow_redirects
request_url, headers=headers, data=data, raise_for_status=True, allow_redirects=self._allow_redirects
) as response,
):
return await response.text()
@@ -206,14 +234,14 @@ class HttpPlugin(KernelBaseModel):
Returns:
The response body as a string.
"""
self._validate_url(url)
request_url = await self._validate_url(url)
headers = {"Content-Type": "application/json"}
data = json.dumps(body) if body is not None else None
async with (
aiohttp.ClientSession() as session,
session.put(
url, headers=headers, data=data, raise_for_status=True, allow_redirects=self._allow_redirects
request_url, headers=headers, data=data, raise_for_status=True, allow_redirects=self._allow_redirects
) as response,
):
return await response.text()
@@ -228,10 +256,10 @@ class HttpPlugin(KernelBaseModel):
Returns:
The response body as a string.
"""
self._validate_url(url)
request_url = await self._validate_url(url)
async with (
aiohttp.ClientSession() as session,
session.delete(url, raise_for_status=True, allow_redirects=self._allow_redirects) as response,
session.delete(request_url, raise_for_status=True, allow_redirects=self._allow_redirects) as response,
):
return await response.text()
python/semantic_kernel/utils/public_network_address_validator.py
+155
View File
@@ -0,0 +1,155 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import ipaddress
import socket
from collections.abc import Awaitable, Callable, Sequence
from urllib.parse import ParseResult
from semantic_kernel.exceptions.function_exceptions import FunctionExecutionException
DnsResolver = Callable[[str], Awaitable[Sequence[str | ipaddress.IPv4Address | ipaddress.IPv6Address]]]
def try_categorize_non_public_address(
address: str | ipaddress.IPv4Address | ipaddress.IPv6Address,
) -> tuple[bool, str]:
"""Return whether an IP address is non-public and the category when blocked."""
ip_address = ipaddress.ip_address(address)
if isinstance(ip_address, ipaddress.IPv6Address) and ip_address.ipv4_mapped:
ip_address = ip_address.ipv4_mapped
if isinstance(ip_address, ipaddress.IPv4Address):
return _try_classify_ipv4(ip_address)
return _try_classify_ipv6(ip_address)
async def ensure_public_host(
parsed_url: ParseResult,
dns_resolver: DnsResolver | None = None,
configuration_hint: str = "",
) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]:
"""Reject non-public destinations and return vetted DNS addresses in resolver order.
The caller is responsible for binding the returned addresses to the connection.
Literal IP hosts are validated but return an empty list because there is no DNS lookup to pin.
"""
host = parsed_url.hostname
if host is None:
raise FunctionExecutionException(f"The request URI '{parsed_url.geturl()}' does not contain a valid host.")
try:
ip_address = ipaddress.ip_address(host)
except ValueError:
addresses = await _resolve_host(host, dns_resolver)
else:
_ensure_public_address(parsed_url.geturl(), ip_address, configuration_hint)
return []
if not addresses:
raise FunctionExecutionException(
f"The request URI '{parsed_url.geturl()}' is not allowed: DNS resolution for host "
f"'{host}' returned no addresses. The request is blocked as a precaution."
)
for address in addresses:
_ensure_public_address(parsed_url.geturl(), address, configuration_hint)
return addresses
async def _resolve_host(
host: str,
dns_resolver: DnsResolver | None,
) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]:
try:
if dns_resolver:
resolved_addresses = await dns_resolver(host)
return [ipaddress.ip_address(address) for address in resolved_addresses]
loop = asyncio.get_running_loop()
addr_info = await loop.getaddrinfo(host, None, type=socket.SOCK_STREAM)
except (OSError, ValueError) as exc:
raise FunctionExecutionException(
f"The request URI host '{host}' is not allowed: DNS resolution failed. "
"The request is blocked as a precaution to prevent potential access to private network addresses."
) from exc
addresses: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = []
seen_addresses: set[str] = set()
for family, _, _, _, sockaddr in addr_info:
if family not in (socket.AF_INET, socket.AF_INET6):
continue
address = ipaddress.ip_address(sockaddr[0])
address_string = str(address)
if address_string not in seen_addresses:
addresses.append(address)
seen_addresses.add(address_string)
return addresses
def _ensure_public_address(
url: str,
address: ipaddress.IPv4Address | ipaddress.IPv6Address,
configuration_hint: str,
) -> None:
blocked, category = try_categorize_non_public_address(address)
if blocked:
raise FunctionExecutionException(
f"The request URI '{url}' is not allowed: host resolves to a {category} address ({address}), "
"which is blocked by default to prevent Server-Side Request Forgery (SSRF). "
f"{configuration_hint}"
)
def _try_classify_ipv4(address: ipaddress.IPv4Address) -> tuple[bool, str]:
b0, b1, b2, _ = address.packed
if b0 == 0:
return True, "unspecified"
if b0 == 10:
return True, "private (RFC1918)"
if b0 == 127:
return True, "loopback"
if b0 == 169 and b1 == 254:
return True, "link-local"
if b0 == 172 and 16 <= b1 <= 31:
return True, "private (RFC1918)"
if b0 == 192 and b1 == 168:
return True, "private (RFC1918)"
if b0 == 100 and 64 <= b1 <= 127:
return True, "carrier-grade NAT"
if b0 == 198 and b1 in (18, 19):
return True, "benchmarking"
if b0 == 192 and b1 == 0 and b2 in (0, 2):
return True, "reserved"
if b0 == 198 and b1 == 51 and b2 == 100:
return True, "reserved"
if b0 == 203 and b1 == 0 and b2 == 113:
return True, "reserved"
if 224 <= b0 <= 239:
return True, "multicast"
if b0 >= 240:
return True, "reserved"
return False, ""
def _try_classify_ipv6(address: ipaddress.IPv6Address) -> tuple[bool, str]:
if address.is_loopback:
return True, "loopback"
if address.is_unspecified:
return True, "unspecified"
if address.is_link_local:
return True, "link-local"
if address.is_site_local:
return True, "site-local"
if address in ipaddress.ip_network("fc00::/7"):
return True, "private (IPv6 ULA)"
if address.is_multicast:
return True, "multicast"
if address in ipaddress.ip_network("2001:db8::/32"):
return True, "reserved"
return False, ""
python/tests/unit/connectors/openapi_plugin/test_openapi_runner_dns_pinning.py
+2 -4
View File
@@ -10,10 +10,8 @@ import httpx
import pytest
from semantic_kernel.connectors.openapi_plugin.openapi_runner import OpenApiRunner
from semantic_kernel.connectors.openapi_plugin.server_url_validator import (
ServerUrlValidationOptions,
try_categorize_non_public_address,
)
from semantic_kernel.connectors.openapi_plugin.server_url_validator import ServerUrlValidationOptions
from semantic_kernel.utils.public_network_address_validator import try_categorize_non_public_address
HOST = "rebind.example"
PUBLIC_ADDRESS = "93.184.216.34"
python/tests/unit/connectors/openapi_plugin/test_server_url_validator.py
+49 -12
View File
@@ -7,10 +7,10 @@ import pytest
from semantic_kernel.connectors.openapi_plugin.server_url_validator import (
ServerUrlValidationOptions,
try_categorize_non_public_address,
validate_server_url,
)
from semantic_kernel.exceptions import FunctionExecutionException
from semantic_kernel.utils.public_network_address_validator import try_categorize_non_public_address
@pytest.mark.parametrize(
@@ -38,8 +38,13 @@ from semantic_kernel.exceptions import FunctionExecutionException
("::1", "loopback"),
("::", "unspecified"),
("fe80::1", "link-local"),
("febf:ffff:ffff:ffff:ffff:ffff:ffff:ffff", "link-local"),
("fec0::", "site-local"),
("fec0::1", "site-local"),
("feff:ffff:ffff:ffff:ffff:ffff:ffff:ffff", "site-local"),
("fc00::1", "private (IPv6 ULA)"),
("fd00::1", "private (IPv6 ULA)"),
("ff00::", "multicast"),
("ff02::1", "multicast"),
("2001:db8::1", "reserved"),
("::ffff:127.0.0.1", "loopback"),
@@ -85,6 +90,12 @@ async def test_validate_server_url_rejects_literal_loopback_ipv6():
await validate_server_url("https://[::1]/")
@pytest.mark.parametrize("address", ["fec0::", "fec0::1", "feff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"])
async def test_validate_server_url_rejects_literal_site_local_ipv6(address):
with pytest.raises(FunctionExecutionException, match="site-local"):
await validate_server_url(f"https://[{address}]/")
async def test_validate_server_url_rejects_http_scheme_by_default():
with pytest.raises(FunctionExecutionException, match="scheme"):
await validate_server_url("http://api.example.com/")
@@ -105,6 +116,15 @@ async def test_validate_server_url_allows_explicit_base_url_for_private_http_add
await validate_server_url("http://192.168.1.100/v1/orders", options)
async def test_validate_server_url_preserves_explicit_hostname_allowlist_dns_bypass():
options = ServerUrlValidationOptions(allowed_base_urls=["http://trusted.example.com/v1"])
async def unexpected_resolver(host: str):
raise AssertionError("An explicitly trusted URL should not be resolved.")
await validate_server_url("http://trusted.example.com/v1/orders", options, dns_resolver=unexpected_resolver)
async def test_validate_server_url_rejects_when_allowed_base_urls_do_not_match():
options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/v1"])
@@ -112,18 +132,28 @@ async def test_validate_server_url_rejects_when_allowed_base_urls_do_not_match()
await validate_server_url("https://api.example.com/v2/orders", options)
async def test_validate_server_url_allows_private_network_access_after_scheme_gate():
@pytest.mark.parametrize("url", ["https://10.0.0.5/", "https://[fec0::1]/"])
async def test_validate_server_url_allows_private_network_access_after_scheme_gate(url):
options = ServerUrlValidationOptions(allow_private_network_access=True)
await validate_server_url("https://10.0.0.5/", options)
await validate_server_url(url, options)
async def test_validate_server_url_blocks_hostname_resolving_to_link_local():
@pytest.mark.parametrize(
("address", "expected_category"),
[
("169.254.169.254", "link-local"),
("fec0::", "site-local"),
("fec0::1", "site-local"),
("feff:ffff:ffff:ffff:ffff:ffff:ffff:ffff", "site-local"),
],
)
async def test_validate_server_url_blocks_hostname_resolving_to_non_public_address(address, expected_category):
async def fake_resolver(host: str):
assert host == "evil.example.com"
return ["169.254.169.254"]
return [address]
with pytest.raises(FunctionExecutionException, match="link-local"):
with pytest.raises(FunctionExecutionException, match=expected_category):
await validate_server_url("https://evil.example.com/latest/meta-data/", dns_resolver=fake_resolver)
@@ -171,14 +201,20 @@ async def test_validate_server_url_blocks_empty_dns_response():
await validate_server_url("https://empty-dns.example.com/", dns_resolver=fake_resolver)
async def test_validate_server_url_returns_validated_addresses_for_pinning():
@pytest.mark.parametrize(
"addresses",
[
["93.184.216.34", "198.41.0.4"],
["2606:4700:4700::1111", "93.184.216.34", "2606:4700:4700::1001"],
],
)
async def test_validate_server_url_returns_validated_addresses_for_pinning(addresses):
async def fake_resolver(host: str):
assert host == "api.example.com"
return ["93.184.216.34", "198.41.0.4"]
return addresses
assert await validate_server_url("https://api.example.com/", dns_resolver=fake_resolver) == [
ipaddress.ip_address("93.184.216.34"),
ipaddress.ip_address("198.41.0.4"),
ipaddress.ip_address(address) for address in addresses
]
@@ -192,8 +228,9 @@ async def test_validate_server_url_returns_validated_ipv6_address_for_pinning():
]
async def test_validate_server_url_returns_no_addresses_for_literal_ip_host():
assert await validate_server_url("https://93.184.216.34/api") == []
@pytest.mark.parametrize("host", ["93.184.216.34", "[2606:4700:4700::1111]"])
async def test_validate_server_url_returns_no_addresses_for_literal_ip_host(host):
assert await validate_server_url(f"https://{host}/api") == []
async def test_validate_server_url_returns_no_addresses_for_allowed_base_url():
python/tests/unit/core_plugins/test_http_plugin.py
+233 -1
View File
@@ -1,9 +1,11 @@
# Copyright (c) Microsoft. All rights reserved.
import logging
from unittest.mock import patch
import socket
from unittest.mock import AsyncMock, patch
import pytest
from yarl import URL
from semantic_kernel import Kernel
from semantic_kernel.core_plugins.http_plugin import HttpPlugin
@@ -11,6 +13,29 @@ from semantic_kernel.exceptions import FunctionExecutionException
from semantic_kernel.functions.kernel_arguments import KernelArguments
def dns_response(*addresses: str):
return [
(
socket.AF_INET6 if ":" in address else socket.AF_INET,
socket.SOCK_STREAM,
socket.IPPROTO_TCP,
"",
(address, 0),
)
for address in addresses
]
@pytest.fixture(autouse=True)
def mock_dns():
with patch(
"asyncio.BaseEventLoop.getaddrinfo",
new_callable=AsyncMock,
return_value=dns_response("93.184.216.34"),
) as resolver:
yield resolver
async def test_it_can_be_instantiated():
plugin = HttpPlugin(allow_all_domains=True)
assert plugin is not None
@@ -443,3 +468,210 @@ async def test_allow_all_domains_with_allowed_domains_allows_redirects():
"""Test that redirects are allowed when both allow_all_domains and allowed_domains are set."""
plugin = HttpPlugin(allowed_domains={"example.com"}, allow_all_domains=True)
assert plugin._is_uri_allowed("https://any-domain.com/path") is True
@pytest.mark.parametrize("method", ["get", "post", "put", "delete"])
@pytest.mark.parametrize(
"address",
[
"169.254.169.254",
"127.0.0.1",
"10.0.0.1",
"172.16.0.1",
"192.168.0.1",
"100.64.0.1",
"0.0.0.0",
"224.0.0.1",
"198.18.0.1",
"192.0.2.1",
"::1",
"fe80::1",
"fec0::",
"fec0::1",
"feff:ffff:ffff:ffff:ffff:ffff:ffff:ffff",
"fc00::1",
"ff02::1",
"::ffff:169.254.169.254",
],
)
async def test_allowed_hostname_resolving_to_non_public_address_is_blocked(method, address, mock_dns):
mock_dns.return_value = dns_response(address)
plugin = HttpPlugin(allowed_domains={"example.com"})
with (
patch("aiohttp.ClientSession") as session,
pytest.raises(FunctionExecutionException, match="host resolves to"),
):
await getattr(plugin, method)("http://example.com/metadata/v1/id")
session.assert_not_called()
mock_dns.assert_awaited_once_with("example.com", None, type=socket.SOCK_STREAM)
@pytest.mark.parametrize("method", ["get", "post", "put", "delete"])
@pytest.mark.parametrize(
"address",
["169.254.169.254", "::1", "::ffff:127.0.0.1", "fec0::", "fec0::1", "feff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"],
)
async def test_allowed_literal_non_public_address_is_blocked_without_dns(method, address, mock_dns):
url_host = f"[{address}]" if ":" in address else address
plugin = HttpPlugin(allowed_domains={address})
with (
patch("aiohttp.ClientSession") as session,
pytest.raises(FunctionExecutionException, match="host resolves to"),
):
await getattr(plugin, method)(f"http://{url_host}/")
session.assert_not_called()
mock_dns.assert_not_awaited()
@pytest.mark.parametrize(
"addresses",
[("93.184.216.34", "10.0.0.1"), ("::1", "2606:4700:4700::1111"), ("93.184.216.34", "fec0::1")],
)
async def test_mixed_public_and_private_dns_response_is_blocked(addresses, mock_dns):
mock_dns.return_value = dns_response(*addresses)
plugin = HttpPlugin(allowed_domains={"example.com"})
with (
patch("aiohttp.ClientSession") as session,
pytest.raises(FunctionExecutionException, match="host resolves to"),
):
await plugin.get("https://example.com/")
session.assert_not_called()
@pytest.mark.parametrize("method", ["get", "post", "put", "delete"])
@pytest.mark.parametrize("failure", ["error", "empty"])
async def test_dns_failure_blocks_all_methods(method, failure, mock_dns):
if failure == "error":
mock_dns.side_effect = socket.gaierror()
else:
mock_dns.return_value = []
plugin = HttpPlugin(allowed_domains={"example.com"})
with (
patch("aiohttp.ClientSession") as session,
pytest.raises(FunctionExecutionException, match="DNS resolution"),
):
await getattr(plugin, method)("https://example.com/")
session.assert_not_called()
@pytest.mark.parametrize("method", ["get", "post", "put", "delete"])
async def test_public_dns_addresses_are_allowed(method, mock_dns):
mock_dns.return_value = dns_response("93.184.216.34", "2606:4700:4700::1111")
plugin = HttpPlugin(allowed_domains={"example.com"})
with patch(f"aiohttp.ClientSession.{method}") as request:
request.return_value.__aenter__.return_value.text.return_value = "public response"
assert await getattr(plugin, method)("http://example.com/") == "public response"
request.assert_called_once_with(URL("http://example.com/"), **request.call_args.kwargs)
assert request.call_args.kwargs["allow_redirects"] is False
mock_dns.assert_awaited_once()
@pytest.mark.parametrize("method", ["get", "post", "put", "delete"])
async def test_idna_hostname_resolving_to_loopback_is_blocked(method, mock_dns):
def resolve(host, *args, **kwargs):
addresses = {"fass.test": "93.184.216.34", "xn--fa-hia.test": "127.0.0.1"}
return dns_response(addresses[host.encode("idna").decode("ascii")])
mock_dns.side_effect = resolve
plugin = HttpPlugin(allowed_domains={"fa\u00df.test"})
with (
patch("aiohttp.ClientSession") as session,
pytest.raises(FunctionExecutionException, match="loopback"),
):
session.return_value.__aenter__.return_value = session.return_value
await getattr(plugin, method)("http://fa\u00df.test/")
session.assert_not_called()
mock_dns.assert_awaited_once_with("xn--fa-hia.test", None, type=socket.SOCK_STREAM)
@pytest.mark.parametrize("method", ["get", "post", "put", "delete"])
@pytest.mark.parametrize("host", ["fa\u00df.test", "xn--fa-hia.test"])
async def test_idna_hostname_is_identical_for_validation_and_request(method, host, mock_dns):
plugin = HttpPlugin(allowed_domains={host})
with patch(f"aiohttp.ClientSession.{method}") as request:
request.return_value.__aenter__.return_value.text.return_value = "public response"
assert await getattr(plugin, method)(f"https://{host}/path") == "public response"
mock_dns.assert_awaited_once_with("xn--fa-hia.test", None, type=socket.SOCK_STREAM)
request.assert_called_once_with(URL("https://xn--fa-hia.test/path"), **request.call_args.kwargs)
assert request.call_args.kwargs["allow_redirects"] is False
async def test_idna_normalization_does_not_bypass_domain_allowlist(mock_dns):
plugin = HttpPlugin(allowed_domains={"fass.test"})
with (
patch("aiohttp.ClientSession") as session,
pytest.raises(FunctionExecutionException, match="Sending requests"),
):
await plugin.get("https://fa\u00df.test/")
session.assert_not_called()
mock_dns.assert_not_awaited()
@pytest.mark.parametrize("address", ["1.1.1.1", "2606:4700:4700::1111"])
async def test_public_literal_address_is_allowed_without_dns(address, mock_dns):
url_host = f"[{address}]" if ":" in address else address
plugin = HttpPlugin(allowed_domains={address})
with patch("aiohttp.ClientSession.get") as request:
request.return_value.__aenter__.return_value.text.return_value = "public response"
assert await plugin.get(f"https://{url_host}/") == "public response"
mock_dns.assert_not_awaited()
@pytest.mark.parametrize("method", ["get", "post", "put", "delete"])
@pytest.mark.parametrize("host", ["example.com", "127.0.0.1", "fec0::1"])
@pytest.mark.parametrize("unrestricted", [False, True])
async def test_private_network_access_requires_explicit_opt_in(method, host, unrestricted, mock_dns):
url_host = f"[{host}]" if ":" in host else host
plugin = HttpPlugin(
allowed_domains={host},
allow_all_domains=unrestricted,
allow_private_network_access=not unrestricted,
)
with patch(f"aiohttp.ClientSession.{method}") as request:
request.return_value.__aenter__.return_value.text.return_value = "internal response"
assert await getattr(plugin, method)(f"http://{url_host}/") == "internal response"
assert request.call_args.kwargs["allow_redirects"] is unrestricted
mock_dns.assert_not_awaited()
@pytest.mark.parametrize("url", ["http://notallowed.com/", "http://example.com:8080/", "file://example.com/"])
async def test_private_network_opt_in_preserves_domain_port_and_scheme_restrictions(url, mock_dns):
plugin = HttpPlugin(allowed_domains={"example.com"}, allow_private_network_access=True)
with (
patch("aiohttp.ClientSession") as session,
pytest.raises(FunctionExecutionException, match="Sending requests"),
):
await plugin.get(url)
session.assert_not_called()
mock_dns.assert_not_awaited()
async def test_denied_domain_is_not_resolved(mock_dns):
plugin = HttpPlugin(allowed_domains={"example.com"})
with pytest.raises(FunctionExecutionException, match="Sending requests"):
await plugin.get("http://notallowed.com/")
mock_dns.assert_not_awaited()