from typing import Dict, Optional, Union from pathlib import Path from urllib.parse import quote import os def _parse_proxy_line(line: str) -> Optional[str]: # Формат: scheme:ip:port[:login[:pass]] # Примеры: # socks5:127.0.0.1:9050 # socks5:127.0.0.1:9050:user:pass # http:127.0.0.1:8888 parts = [p.strip() for p in line.strip().split(":")] if len(parts) < 3: return None scheme, host, port = parts[0], parts[1], parts[2] user = parts[3] if len(parts) >= 4 and parts[3] else None password = parts[4] if len(parts) >= 5 and parts[4] else None auth = "" if user: auth = quote(user) if password: auth += f":{quote(password)}" auth += "@" # Исправление для socks5: httpx ожидает схему socks5:// (не socks://) if scheme == "socks": scheme = "socks5" # Явно проверяем протокол, чтобы был http://, https:// или socks5:// if not scheme.startswith(("http", "socks")): scheme = "http" return f"{scheme}://{auth}{host}:{port}" def _read_proxy_from_file() -> Optional[str]: file_path = Path("proxy.txt") if not file_path.exists(): return None try: for raw in file_path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#"): continue # поддержим дополнительные ключи вида key=value в этом же файле (разберём ниже) if "=" in line: continue url = _parse_proxy_line(line) if url: return url except Exception: return None return None def build_httpx_proxies() -> Optional[Dict[str, str]]: # Читаем только из proxy.txt (без переменных окружения) url = _read_proxy_from_file() if not url: return None # Для httpx корректнее указывать схемы явно return { "http://": url, "https://": url, } def _read_kv_from_proxy_file() -> Dict[str, str]: """ Поддержка дополнительных опций в proxy.txt: ca=/полный/путь/к/burp-ca.pem verify=false # отключить проверку сертификатов (для отладки) """ out: Dict[str, str] = {} p = Path("proxy.txt") if not p.exists(): return out try: for raw in p.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#"): continue if "=" not in line: continue k, v = line.split("=", 1) out[k.strip().lower()] = v.strip() except Exception: return out return out def get_tls_verify() -> Union[bool, str]: """ Возвращает значение для параметра httpx.AsyncClient(verify=...): - путь к PEM-бандлу (строка), если нашли ca=... или файл proxy-ca.pem в корне - False, если verify=false/insecure=1/AGENTUI_VERIFY=false - True по умолчанию - Новое: можно задать флаг второй «голой» строкой в proxy.txt (после URL прокси): пример: http:127.0.0.1:8888 false или http:127.0.0.1:8888 true """ # 1) Переменные окружения имеют приоритет env_verify = os.getenv("AGENTUI_VERIFY") if env_verify is not None and env_verify.strip().lower() in ("0", "false", "no", "off"): return False env_ca = os.getenv("AGENTUI_CA") if env_ca: path = Path(env_ca).expanduser() if path.exists(): return str(path) # 2) proxy.txt ключи kv = _read_kv_from_proxy_file() if kv.get("verify", "").lower() in ("0", "false", "no", "off"): return False if "ca" in kv: path = Path(kv["ca"]).expanduser() if path.exists(): return str(path) # 2.1) Дополнительно: поддержка второй строки без ключа — true/false try: p = Path("proxy.txt") if p.exists(): lines = [ln.strip() for ln in p.read_text(encoding="utf-8").splitlines()] # найдём первую «URL» строку (без '=' и не пустую/коммент) idx_url = -1 for i, ln in enumerate(lines): if not ln or ln.startswith("#") or "=" in ln: continue idx_url = i break if idx_url >= 0: # ищем следующую «голую» строку for j in range(idx_url + 1, len(lines)): ln = lines[j].strip() if not ln or ln.startswith("#") or "=" in ln: continue low = ln.lower() if low in ("1", "true", "yes", "on"): return True if low in ("0", "false", "no", "off"): return False # если это не похожее на флаг, игнорируем и продолжаем except Exception: pass # 3) Файл по умолчанию в корне проекта default_ca = Path("proxy-ca.pem") if default_ca.exists(): return str(default_ca) # 4) По умолчанию строгая проверка return True def is_verify_explicit() -> bool: """ Возвращает True, если пользователь ЯВНО задал политику проверки TLS, чтобы клиент не переопределял её значением по умолчанию. Учитываются: - переменные окружения: AGENTUI_VERIFY, AGENTUI_CA - ключи в proxy.txt: verify=..., ca=... - файл proxy-ca.pem в корне проекта - Новое: «вторая голая строка» после URL в proxy.txt со значением true/false """ if os.getenv("AGENTUI_VERIFY") is not None: return True if os.getenv("AGENTUI_CA"): return True kv = _read_kv_from_proxy_file() if "verify" in kv or "ca" in kv: return True # Вторая «голая» строка как явный флаг try: p = Path("proxy.txt") if p.exists(): lines = [ln.strip() for ln in p.read_text(encoding="utf-8").splitlines()] idx_url = -1 for i, ln in enumerate(lines): if not ln or ln.startswith("#") or "=" in ln: continue idx_url = i break if idx_url >= 0: for j in range(idx_url + 1, len(lines)): ln = lines[j].strip() if not ln or ln.startswith("#") or "=" in ln: continue if ln.lower() in ("1", "0", "true", "false", "yes", "no", "on", "off"): return True break except Exception: pass if Path("proxy-ca.pem").exists(): return True return False