Recipe for preventing SSRF?

[DavidBuchanan314]

I'm writing a web application that accepts user-provided URLs, and queries them using an aiohttp client.

I want to restrict this client from being able to access "internal" network resources, including but not limited to "localhost". I want it to remain restricted even if it follows HTTP redirects.

Some more background on why this matters: https://portswigger.net/web-security/ssrf

Does anyone have some ideas for how to approach this?

[DavidBuchanan314]

I'm thinking of making a custom subclass of TCPConnector, and wrapping _resolve_host to filter out any internal IP addresses

[Dreamsorcerer]

That might work for now (though obviously we don't guarantee that we won't break it in future). A safer approach for now, might be to disable redirects and then check the URL each time. In future, we plan to add a generic middlewares to the client, which will likely provide a simple mechanism to do this reliably. #6915

[DavidBuchanan314]

Thank you for the response. Unfortunately checking the URL itself isn't very safe, since someone could configure a domain that resolves to 127.0.0.1 (for example), and for my use-case I do need to follow redirects. I guess I will bodge TCPConnector and pin to a specific version of aiohttp to avoid breakage, and hopefully upgrade once an alternative approach is available (or until I think of something better heh).

[Dreamsorcerer]

Yes, I meant handling the redirects yourself, rather than not supporting them.
But, as you mention, the DNS resolution needs to happen, so not a good solution.

[DavidBuchanan314]

here's my temporary solution: (it's fragile since it relies on private method _convert_hosts_to_addr_infos)

import ipaddress
from aiohttp import TCPConnector, ClientSession
import asyncio

# not a public API, at least check if it exists
assert hasattr(TCPConnector, "_convert_hosts_to_addr_infos")


class SSRFException(ValueError):
	pass


class SSRFProtectedTCPConnector(TCPConnector):
	def _convert_hosts_to_addr_infos(self, hosts):
		for host in hosts:
			if ipaddress.ip_address(host["host"]).is_private:
				raise SSRFException(
					"Can't connect to private IP: " + host["host"]
				)
		return super()._convert_hosts_to_addr_infos(hosts)


async def main():
	async with ClientSession(connector=SSRFProtectedTCPConnector()) as client:
		# normal requests work
		async with client.get("http://example.com/") as resp:
			assert resp.ok

		# private IPs are blocked
		for testcase in [
			"http://192.168.0.1/",
			"http://localhost/",
			"https://localhost/",
			"http://127.0.0.1/",
			"https://tinyurl.com/localssrftest",  # works for redirects, too!
		]:
			try:
				async with client.get(testcase) as resp:
					print(resp.status)
				assert False, "should have raised an exception"
			except SSRFException as e:
				print(repr(e))


if __name__ == "__main__":
	asyncio.run(main())

[bdraco]

ipaddress.ip_address(...).is_private

That check might be slow since constructing ip_address objects are slow.

I'd wrap it in an lru_cache to check if the string is a private ip

[DavidBuchanan314]

I came up with a slightly cleaner method, which is to provide a custom DNS resolver (wrapping the default one) which filters its responses. However, it requires a monkeypatch to stop TCPConnector from taking a shortcut in resolving bare IPs (otherwise they'd never pass through the Resolver to get detected). Would you consider a PR that adds some kind of always_resolve flag to TCPConnector, letting me avoid the monkeypatch?

aiohttp/aiohttp/connector.py

Lines 908 to 915 in 45de81d

	async def _resolve_host(
	self, host: str, port: int, traces: Optional[Sequence["Trace"]] = None
	) -> List[ResolveResult]:
	"""Resolve host and return list of addresses."""
	if is_ip_address(host):
	return [
	{
	"hostname": host,

import ipaddress
from aiohttp import TCPConnector, ClientSession
import aiohttp.connector
from aiohttp.resolver import DefaultResolver, AbstractResolver
import asyncio

# XXX: monkeypatch to force all hosts to go through the resolver
aiohttp.connector.is_ip_address = lambda _: False

class SSRFException(ValueError):
	pass

class SSRFSafeResolverWrapper(AbstractResolver):
	def __init__(self, resolver: AbstractResolver):
		self.resolver = resolver

	async def resolve(self, host: str, port: int, family: int):
		result = await self.resolver.resolve(host, port, family)
		for host in result:
			if ipaddress.ip_address(host["host"]).is_private:
				raise SSRFException("Can't connect to private IP: " + host["host"])
		return result
	
	async def close(self) -> None:
		await self.resolver.close()

async def main():
	resolver = SSRFSafeResolverWrapper(DefaultResolver())
	connector = TCPConnector(resolver=resolver)
	async with ClientSession(connector=connector) as client:
		# normal requests work
		async with client.get("http://example.com/") as resp:
			assert resp.ok

		# private IPs are blocked
		for testcase in [
			"http://localhost/",
			"http://192.168.0.1/",
			"https://localhost/",
			"http://127.0.0.1/",
			"https://tinyurl.com/localssrftest",  # works for redirects, too!
		]:
			try:
				async with client.get(testcase) as resp:
					print(resp.status)
				assert False, "should have raised an exception"
			except SSRFException as e:
				print(repr(e))


if __name__ == "__main__":
	asyncio.run(main())

[Dreamsorcerer]

Or do you anticipate the middlewares having a step for DNS resolution somehow?

My original thought was roughly like:

async def middleware(request, handler):
    # Can check if request is to a private IP here, unresolved only.
    await handler(request)

If we wanted to include DNS resolution it'd be more complex. I'm struggling to think of a middleware design that allows code to run pre and post-resolution.

[Dreamsorcerer]

Having 2 functions could work, but probably makes it far more likely that someone will mess up a middleware. Also, not sure how many cases there would be where this actually matters, so I suspect it's not worth the extra complexity.

async def middleware(raw_request, resolver, handler):
    request = await resolver(raw_request)
    # Check resolved IP is a private IP here.
    await handler(request)

[bdraco]

Or do you anticipate the middlewares having a step for DNS resolution somehow?

My original thought was roughly like:

async def middleware(request, handler):

    # Can check if request is to a private IP here, unresolved only.

    await handler(request)

If we wanted to include DNS resolution it'd be more complex. I'm struggling to think of a middleware design that allows code to run pre and post-resolution.

I wasn't thinking resolution would be included but I guess we would need to do so for this case so many not a good fit after all.

[Dreamsorcerer]

Maybe having both middleware and custom resolver would be a decent solution. Then we could probably put that into a library to ensure users don't have to figure out how to do it correctly (e.g. aiohttp-csrf).

[DavidBuchanan314]

When I first started looking into this I was a little surprised not to find an aiohttp-ssrf already - I'd be happy to contribute that in the future