80 lines
4.8 KiB
Python
80 lines
4.8 KiB
Python
import socket
|
||
import unittest
|
||
from unittest.mock import patch
|
||
|
||
from plugin_loader import plugin
|
||
from chicktools.utils.callback_policy import ALLOWED_CALLBACK_HOSTS
|
||
from chicktools.utils.url_security import CallbackURLValidationError, validate_callback_url
|
||
|
||
|
||
def answer(address):
|
||
return (socket.AF_INET6 if ":" in address else socket.AF_INET, socket.SOCK_STREAM, 6, "", (address, 443))
|
||
|
||
|
||
class URLTests(unittest.TestCase):
|
||
def test_only_reviewed_host_is_allowed(self):
|
||
self.assertEqual(ALLOWED_CALLBACK_HOSTS, frozenset({"notify.bro3.net"}))
|
||
for host in ("notify.bro3.net", "NOTIFY.BRO3.NET", "notify.bro3.net."):
|
||
for scheme in ("https",):
|
||
with self.subTest(host=host, scheme=scheme), patch("socket.getaddrinfo", return_value=[answer("8.8.8.8")]):
|
||
target = validate_callback_url(f"{scheme}://{host}:8443/path?q=1")
|
||
self.assertEqual(target.hostname, "notify.bro3.net")
|
||
self.assertEqual(target.url, f"{scheme}://notify.bro3.net:8443/path?q=1")
|
||
|
||
def test_unlisted_hosts_are_rejected_before_dns(self):
|
||
hosts = ("callback.example", "bro3.net", "a.notify.bro3.net", "notify.bro3.net.evil.example",
|
||
"evilnotify.bro3.net", "notify-bro3.net", "notify.bro3.net..", "notify%2ebro3.net",
|
||
"notifу.bro3.net", "8.8.8.8", "[2606:4700:4700::1111]", "2130706433")
|
||
for host in hosts:
|
||
with self.subTest(host=host), patch("socket.getaddrinfo") as resolver:
|
||
with self.assertRaises(CallbackURLValidationError):
|
||
validate_callback_url("https://" + host + "/callback")
|
||
resolver.assert_not_called()
|
||
|
||
def test_loopback_and_private_literals_are_rejected(self):
|
||
for host in ("127.0.0.1", "localhost", "localhost.", "a.localhost", "10.0.0.1", "172.16.0.1",
|
||
"192.168.1.1", "169.254.169.254", "100.64.0.1", "198.18.0.1", "0.0.0.0",
|
||
"224.0.0.1", "[::1]", "[::]", "[fc00::1]", "[fe80::1]", "[::ffff:127.0.0.1]",
|
||
"[64:ff9b::7f00:1]", "metadata.google.internal"):
|
||
with self.subTest(host=host), patch("socket.getaddrinfo") as resolver:
|
||
with self.assertRaises(CallbackURLValidationError):
|
||
validate_callback_url("https://" + host)
|
||
resolver.assert_not_called()
|
||
|
||
def test_malformed_urls(self):
|
||
for url in ("ftp://notify.bro3.net", "file:///x", "gopher://notify.bro3.net", "http://",
|
||
"https://notify.bro3.net:0", "https://notify.bro3.net:99999", "https://notify.bro3.net:wrong",
|
||
"http://[broken", "https://user:pass@notify.bro3.net", "https://notify.bro3.net@evil.example",
|
||
"https://notify.bro3.net/#fragment", "https://notify.bro3.net/a\nb", "https://notify.bro3.net\\foo"):
|
||
with self.subTest(url=url), self.assertRaises(CallbackURLValidationError):
|
||
validate_callback_url(url)
|
||
|
||
def test_allowed_host_resolving_to_private_or_special_address_is_rejected(self):
|
||
for ip in ("127.0.0.1", "10.0.0.1", "169.254.169.254", "100.64.0.1", "198.18.0.1",
|
||
"224.0.0.1", "::1", "fc00::1", "fe80::1", "::ffff:127.0.0.1", "64:ff9b::7f00:1"):
|
||
with self.subTest(ip=ip), patch("socket.getaddrinfo", return_value=[answer(ip)]):
|
||
with self.assertRaisesRegex(CallbackURLValidationError, "public addresses"):
|
||
validate_callback_url("https://notify.bro3.net/callback")
|
||
|
||
def test_dns_mixed_private_answers_rejected(self):
|
||
with patch("socket.getaddrinfo", return_value=[answer("8.8.8.8"), answer("127.0.0.1")]):
|
||
with self.assertRaisesRegex(CallbackURLValidationError, "public addresses"):
|
||
validate_callback_url("https://notify.bro3.net")
|
||
|
||
def test_dns_failure_and_empty_answers_fail_closed(self):
|
||
with patch("socket.getaddrinfo", side_effect=socket.gaierror("failed")):
|
||
with self.assertRaisesRegex(CallbackURLValidationError, "DNS resolution failed"):
|
||
validate_callback_url("https://notify.bro3.net")
|
||
with patch("socket.getaddrinfo", return_value=[]):
|
||
with self.assertRaisesRegex(CallbackURLValidationError, "public addresses"):
|
||
validate_callback_url("https://notify.bro3.net")
|
||
|
||
def test_dns_resolved_once_and_saved(self):
|
||
for address in ("8.8.8.8", "2606:4700:4700::1111"):
|
||
with patch("socket.getaddrinfo", return_value=[answer(address)]) as resolver:
|
||
target = validate_callback_url("https://notify.bro3.net/finished?q=1")
|
||
resolver.assert_called_once()
|
||
self.assertEqual(target.address, address)
|
||
self.assertEqual(target.hostname, "notify.bro3.net")
|
||
self.assertEqual(target.url, "https://notify.bro3.net/finished?q=1")
|