from  future  import annotations

import ipaddress
import logging
import os
import socket
import urllib.parse
import urllib.request
from typing import Optional, Sequence, Union
import requests

log = logging.getLogger( name )

# Same text as open webui.constants.ERROR MESSAGES.INVALID URL
INVALID URL = (
  "Oops! The URL you provided is invalid. Please double-check and try again."
)

# Same semantics as open webui.config (ENABLE RAG LOCAL WEB FETCH / WEB FETCH FILTER LIST)
ENABLE RAG LOCAL WEB FETCH = (
  os.getenv("ENABLE RAG LOCAL WEB FETCH", "False").lower() == "true"
)

 DEFAULT WEB FETCH FILTER LIST = [
  "!169.254.169.254",
  "!fd00:ec2::254",
  "!metadata.google.internal",
  "!metadata.azure.com",
  "!100.100.100.200",
]
 web fetch filter env = os.getenv("WEB FETCH FILTER LIST", "")
if web fetch filter env == "":
   web fetch filter env list: list[str] = []
else:
   web fetch filter env list = [
    item.strip()
    for item in web fetch filter env.split(",")
    if item.strip()
  ]
WEB FETCH FILTER LIST = list(
  set( DEFAULT WEB FETCH FILTER LIST + web fetch filter env list)
)


def get allow block lists(filter list):
  allow list = []
  block list = []

  if filter list:
    for d in filter list:
      if d.startswith("!"):
        block list.append(d[1:].strip())
      else:
        allow list.append(d.strip())

  return allow list, block list


def is string allowed(
  string: Union[str, Sequence[str]], filter list: Optional[list[str]] = None
) -> bool:
  if not filter list:
    return True

  allow list, block list = get allow block lists(filter list)
  strings = [string] if isinstance(string, str) else list(string)

  if allow list:
    if not any(s.endswith(allowed) for s in strings for allowed in allow list):
      return False

  if any(s.endswith(blocked) for s in strings for blocked in block list):
    return False

  return True


def resolve hostname(hostname):
  # Get address information
  addr info = socket.getaddrinfo(hostname, None)

  # Extract IP addresses from address information
  ipv4 addresses = [info[4][0] for info in addr info if info[0] == socket.AF INET]
  ipv6 addresses = [info[4][0] for info in addr info if info[0] == socket.AF INET6]

  return ipv4 addresses, ipv6 addresses


def validators url accept(url: str) -> bool:
  """
  Stand-in for python-validators url(): True if string looks like http(s) URL with host.
  """
  try:
    u = url.strip()
    if not u:
      return False
    p = urllib.parse.urlparse(u)
    if p.scheme not in ("http", "https"):
      return False
    if not p.netloc:
      return False
    return True
  except Exception:
    return False


def ipv4 private(ip: str) -> bool:
  try:
    a = ipaddress.ip address(ip)
    return a.version == 4 and a.is private
  except ValueError:
    return False


def ipv6 private(ip: str) -> bool:
  try:
    a = ipaddress.ip address(ip)
    return a.version == 6 and a.is private
  except ValueError:
    return False


def validate url(url: Union[str, Sequence[str]]):
  if isinstance(url, str):
    if not validators url accept(url):
      raise ValueError(INVALID URL)

    parsed url = urllib.parse.urlparse(url)

    # Protocol validation - only allow http/https
    if parsed url.scheme not in ["http", "https"]:
      log.warning(
        f"Blocked non-HTTP(S) protocol: {parsed url.scheme} in URL: {url}"
      )
      raise ValueError(INVALID URL)

    # Blocklist check using unified filtering logic
    if WEB FETCH FILTER LIST:
      if not is string allowed(url, WEB FETCH FILTER LIST):
        log.warning(f"URL blocked by filter list: {url}")
        raise ValueError(INVALID URL)

    if not ENABLE RAG LOCAL WEB FETCH:
      # Local web fetch is disabled, filter out any URLs that resolve to private IP addresses
      parsed url = urllib.parse.urlparse(url)
      # Get IPv4 and IPv6 addresses
      ipv4 addresses, ipv6 addresses = resolve hostname(parsed url.hostname)
      # Check if any of the resolved addresses are private
      # This is technically still vulnerable to DNS rebinding attacks, as we don't control WebBaseLoader
      for ip in ipv4 addresses:
        if ipv4 private(ip):
          raise ValueError(INVALID URL)
      for ip in ipv6 addresses:
        if ipv6 private(ip):
          raise ValueError(INVALID URL)
    return True
  elif isinstance(url, Sequence):
    return all(validate url(u) for u in url)
  else:
    return False

if  name  == " main ":
  logging.basicConfig(level=logging.INFO)
  # url = "https://127.0.0.1:6666@1.1.1.1"
  url = "https://127.0.0.1:6666"
  validate url(url)
  response = requests.get(url)
  print(response.text)

http://127.0.0.1:6666@baidu.com