fix: enhance URL validation to prevent access to blocked internal addresses

Author: liuruibin

apps/common/utils/fork.py

@@ -1,5 +1,7 @@
 import copy
+import ipaddress
 import re
+import socket
 import traceback
 from functools import reduce
 from typing import List, Set
@@ -13,6 +15,44 @@
 
 requests.packages.urllib3.disable_warnings()
 
+_BLOCKED_NETWORKS = [
+    ipaddress.ip_network("0.0.0.0/8"),
+    ipaddress.ip_network("10.0.0.0/8"),
+    ipaddress.ip_network("100.64.0.0/10"),
+    ipaddress.ip_network("127.0.0.0/8"),
+    ipaddress.ip_network("169.254.0.0/16"),
+    ipaddress.ip_network("172.16.0.0/12"),
+    ipaddress.ip_network("192.168.0.0/16"),
+    ipaddress.ip_network("198.18.0.0/15"),
+    ipaddress.ip_network("198.51.100.0/24"),
+    ipaddress.ip_network("203.0.113.0/24"),
+    ipaddress.ip_network("240.0.0.0/4"),
+    ipaddress.ip_network("::1/128"),
+    ipaddress.ip_network("fc00::/7"),
+    ipaddress.ip_network("fe80::/10"),
+]
+
+
+def _validate_url(url: str) -> None:
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        raise ValueError(f"Disallowed URL scheme: {parsed.scheme!r}")
+    hostname = parsed.hostname
+    if not hostname:
+        raise ValueError("Missing hostname in URL")
+    try:
+        addr_infos = socket.getaddrinfo(hostname, None)
+    except socket.gaierror as exc:
+        raise ValueError(f"Cannot resolve host {hostname!r}: {exc}") from exc
+    for addr_info in addr_infos:
+        ip_str = addr_info[4][0]
+        try:
+            ip = ipaddress.ip_address(ip_str)
+        except ValueError:
+            continue
+        if any(ip in net for net in _BLOCKED_NETWORKS):
+            raise ValueError(f"URL resolves to a blocked internal address: {ip}")
+
 
 class ChildLink:
     def __init__(self, url, tag):
@@ -29,27 +69,34 @@ def fork(self, level: int, exclude_link_url: Set[str], fork_handler):
         self.fork_child(ChildLink(self.base_url, None), self.selector_list, level, exclude_link_url, fork_handler)
 
     @staticmethod
-    def fork_child(child_link: ChildLink, selector_list: List[str], level: int, exclude_link_url: Set[str],
-                   fork_handler):
+    def fork_child(
+        child_link: ChildLink, selector_list: List[str], level: int, exclude_link_url: Set[str], fork_handler
+    ):
         if level < 0:
             return
         else:
             child_link.url = remove_fragment(child_link.url)
-            child_url = child_link.url[:-1] if child_link.url.endswith('/') else child_link.url
+            child_url = child_link.url[:-1] if child_link.url.endswith("/") else child_link.url
         if not exclude_link_url.__contains__(child_url):
             exclude_link_url.add(child_url)
             response = Fork(child_link.url, selector_list).fork()
             fork_handler(child_link, response)
             for child_link in response.child_link_list:
-                child_url = child_link.url[:-1] if child_link.url.endswith('/') else child_link.url
+                child_url = child_link.url[:-1] if child_link.url.endswith("/") else child_link.url
                 if not exclude_link_url.__contains__(child_url):
                     ForkManage.fork_child(child_link, selector_list, level - 1, exclude_link_url, fork_handler)
 
 
 def remove_fragment(url: str) -> str:
     parsed_url = urlparse(url)
-    modified_url = ParseResult(scheme=parsed_url.scheme, netloc=parsed_url.netloc, path=parsed_url.path,
-                               params=parsed_url.params, query=parsed_url.query, fragment=None)
+    modified_url = ParseResult(
+        scheme=parsed_url.scheme,
+        netloc=parsed_url.netloc,
+        path=parsed_url.path,
+        params=parsed_url.params,
+        query=parsed_url.query,
+        fragment=None,
+    )
     return urlunparse(modified_url)
 
 
@@ -63,54 +110,68 @@ def __init__(self, content: str, child_link_list: List[ChildLink], status, messa
 
         @staticmethod
         def success(html_content: str, child_link_list: List[ChildLink]):
-            return Fork.Response(html_content, child_link_list, 200, '')
+            return Fork.Response(html_content, child_link_list, 200, "")
 
         @staticmethod
         def error(message: str):
-            return Fork.Response('', [], 500, message)
+            return Fork.Response("", [], 500, message)
 
     def __init__(self, base_fork_url: str, selector_list: List[str]):
+        _validate_url(base_fork_url)
         base_fork_url = remove_fragment(base_fork_url)
         parsed = urlparse(base_fork_url)
-        path = parsed.path.rstrip('/')
-        self.base_fork_url = urlunparse((
-            parsed.scheme,
-            parsed.netloc,
-            path,
-            None,
-            None,
-            None  # fragment
-        ))
+        path = parsed.path.rstrip("/")
+        self.base_fork_url = urlunparse(
+            (
+                parsed.scheme,
+                parsed.netloc,
+                path,
+                None,
+                None,
+                None,  # fragment
+            )
+        )
         parsed = urlsplit(base_fork_url)
         query = parsed.query
         if query is not None and len(query) > 0:
-            self.base_fork_url = self.base_fork_url + '?' + query
+            self.base_fork_url = self.base_fork_url + "?" + query
         self.selector_list = [selector for selector in selector_list if selector is not None and len(selector) > 0]
         self.urlparse = urlparse(self.base_fork_url)
-        self.base_url = ParseResult(scheme=self.urlparse.scheme, netloc=self.urlparse.netloc, path='', params='',
-                                    query='',
-                                    fragment='').geturl()
+        self.base_url = ParseResult(
+            scheme=self.urlparse.scheme, netloc=self.urlparse.netloc, path="", params="", query="", fragment=""
+        ).geturl()
 
     def get_child_link_list(self, bf: BeautifulSoup):
         # Compute the crawl prefix: parent directory when base_fork_url is an HTML file
         crawl_prefix = self.base_fork_url
-        if crawl_prefix.endswith(('.html', '.htm')):
-            crawl_prefix = crawl_prefix.rsplit('/', 1)[0]
+        if crawl_prefix.endswith((".html", ".htm")):
+            crawl_prefix = crawl_prefix.rsplit("/", 1)[0]
         pattern = "^((?!(http:|https:|tel:/|#|mailto:|javascript:))|" + crawl_prefix + "|/).*"
-        link_list = bf.find_all(name='a', href=re.compile(pattern))
-        result = [ChildLink(link.get('href'), link) if link.get('href').startswith(self.base_url) else ChildLink(
-            self.base_url + link.get('href'), link) for link in link_list]
+        link_list = bf.find_all(name="a", href=re.compile(pattern))
+        result = [
+            ChildLink(link.get("href"), link)
+            if link.get("href").startswith(self.base_url)
+            else ChildLink(self.base_url + link.get("href"), link)
+            for link in link_list
+        ]
         result = [row for row in result if row.url.startswith(crawl_prefix)]
         return result
 
     def get_content_html(self, bf: BeautifulSoup):
         if self.selector_list is None or len(self.selector_list) == 0:
             return str(bf)
-        params = reduce(lambda x, y: {**x, **y},
-                        [{'class_': selector.replace('.', '')} if selector.startswith('.') else
-                         {'id': selector.replace("#", "")} if selector.startswith("#") else {'name': selector} for
-                         selector in
-                         self.selector_list], {})
+        params = reduce(
+            lambda x, y: {**x, **y},
+            [
+                {"class_": selector.replace(".", "")}
+                if selector.startswith(".")
+                else {"id": selector.replace("#", "")}
+                if selector.startswith("#")
+                else {"name": selector}
+                for selector in self.selector_list
+            ],
+            {},
+        )
         f = bf.find_all(**params)
         return "\n".join([str(row) for row in f])
 
@@ -119,53 +180,58 @@ def reset_url(tag, field, base_fork_url):
         field_value: str = tag[field]
         if field_value.startswith("/"):
             result = urlparse(base_fork_url)
-            result_url = ParseResult(scheme=result.scheme, netloc=result.netloc, path=field_value, params='', query='',
-                                     fragment='').geturl()
+            result_url = ParseResult(
+                scheme=result.scheme, netloc=result.netloc, path=field_value, params="", query="", fragment=""
+            ).geturl()
         else:
             # When base_fork_url is an HTML file (not a directory), resolve relative
             # links against its parent directory to avoid broken paths like
             # /en/index.html/about_dolphindb.html
-            if base_fork_url.endswith(('.html', '.htm')):
-                base = base_fork_url.rsplit('/', 1)[0] + '/'
+            if base_fork_url.endswith((".html", ".htm")):
+                base = base_fork_url.rsplit("/", 1)[0] + "/"
             else:
-                base = base_fork_url + '/'
+                base = base_fork_url + "/"
             result_url = urljoin(base, field_value)
-        result_url = result_url[:-1] if result_url.endswith('/') else result_url
+        result_url = result_url[:-1] if result_url.endswith("/") else result_url
         tag[field] = result_url
 
     def reset_beautiful_soup(self, bf: BeautifulSoup):
         reset_config_list = [
             {
-                'field': 'href',
+                "field": "href",
             },
             {
-                'field': 'src',
-            }
+                "field": "src",
+            },
         ]
         for reset_config in reset_config_list:
-            field = reset_config.get('field')
-            tag_list = bf.find_all(**{field: re.compile('^(?!(http:|https:|tel:/|#|mailto:|javascript:)).*')})
+            field = reset_config.get("field")
+            tag_list = bf.find_all(**{field: re.compile("^(?!(http:|https:|tel:/|#|mailto:|javascript:)).*")})
             for tag in tag_list:
                 self.reset_url(tag, field, self.base_fork_url)
             # 去掉 href 以 # 开头的锚点链接，保留文字
-        for a in bf.find_all('a', href=re.compile('^#')):
+        for a in bf.find_all("a", href=re.compile("^#")):
             a.unwrap()
         return bf
 
     @staticmethod
     def get_beautiful_soup(response):
-        encoding = response.encoding if response.encoding is not None and response.encoding != 'ISO-8859-1' else response.apparent_encoding
+        encoding = (
+            response.encoding
+            if response.encoding is not None and response.encoding != "ISO-8859-1"
+            else response.apparent_encoding
+        )
         html_content = response.content.decode(encoding)
         beautiful_soup = BeautifulSoup(html_content, "html.parser")
-        meta_list = beautiful_soup.find_all('meta')
+        meta_list = beautiful_soup.find_all("meta")
         charset_list = Fork.get_charset_list(meta_list)
         if len(charset_list) > 0:
             charset = charset_list[0]
             if charset != encoding:
                 try:
-                    html_content = response.content.decode(charset, errors='replace')
+                    html_content = response.content.decode(charset, errors="replace")
                 except Exception as e:
-                    maxkb_logger.error(f'{e}: {traceback.format_exc()}')
+                    maxkb_logger.error(f"{e}: {traceback.format_exc()}")
                 return BeautifulSoup(html_content, "html.parser")
         return beautiful_soup
 
@@ -174,39 +240,50 @@ def get_charset_list(meta_list):
         charset_list = []
         for meta in meta_list:
             if meta.attrs is not None:
-                if 'charset' in meta.attrs:
-                    charset_list.append(meta.attrs.get('charset'))
-                elif meta.attrs.get('http-equiv', '').lower() == 'content-type' and 'content' in meta.attrs:
-                    match = re.search(r'charset=([^\s;]+)', meta.attrs['content'], re.I)
+                if "charset" in meta.attrs:
+                    charset_list.append(meta.attrs.get("charset"))
+                elif meta.attrs.get("http-equiv", "").lower() == "content-type" and "content" in meta.attrs:
+                    match = re.search(r"charset=([^\s;]+)", meta.attrs["content"], re.I)
                     if match:
                         charset_list.append(match.group(1))
         return charset_list
 
     def fork(self):
         try:
-
             headers = {
-                'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.0.4844.51 Safari/537.36'
+                "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.0.4844.51 Safari/537.36"
             }
 
-            maxkb_logger.info(f'fork:{self.base_fork_url}')
-            response = requests.get(self.base_fork_url, verify=False, headers=headers)
+            maxkb_logger.info(f"fork:{self.base_fork_url}")
+            url = self.base_fork_url
+            for _ in range(10):
+                response = requests.get(url, verify=True, headers=headers, allow_redirects=False, timeout=(10, 30))
+                if response.status_code in (301, 302, 303, 307, 308):
+                    location = response.headers.get("Location")
+                    if not location:
+                        break
+                    location = urljoin(url, location)
+                    _validate_url(location)
+                    url = location
+                    continue
+                break
             if response.status_code != 200:
                 maxkb_logger.error(f"url: {self.base_fork_url} code:{response.status_code}")
                 return Fork.Response.error(f"url: {self.base_fork_url} code:{response.status_code}")
             bf = self.get_beautiful_soup(response)
         except Exception as e:
-            maxkb_logger.error(f'{str(e)}:{traceback.format_exc()}')
+            maxkb_logger.error(f"{str(e)}:{traceback.format_exc()}")
             return Fork.Response.error(str(e))
         bf = self.reset_beautiful_soup(bf)
         link_list = self.get_child_link_list(bf)
         content = self.get_content_html(bf)
 
-        r = markdownify(content, heading_style='ATX')
+        r = markdownify(content, heading_style="ATX")
         return Fork.Response.success(r, link_list)
 
 
 def handler(base_url, response: Fork.Response):
     maxkb_logger.info(base_url.url, base_url.tag.text if base_url.tag else None, response.content)
 
+
 # ForkManage('https://bbs.fit2cloud.com/c/de/6', ['.md-content']).fork(3, set(), handler)