HadTavern/agentui/config.py

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 _read_second_bare_flag_from_proxy() -> Optional[bool]:
    """
    Читает «вторую голую строку» после URL в proxy.txt и интерпретирует как флаг verify:
      true/1/yes/on  -> True
      false/0/no/off -> False
    Возвращает None, если строка отсутствует или не распознана.
    """
    try:
        p = Path("proxy.txt")
        if not p.exists():
            return None
        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
                # если это не похожее на флаг — считаем отсутствующим
                break
    except Exception:
        return None
    return None
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
    second = _read_second_bare_flag_from_proxy()
    if second is True:
        return True
    if second is False:
        return False

    # 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
    # Вторая «голая» строка как явный флаг
    second = _read_second_bare_flag_from_proxy()
    if second is not None:
        return True

    if Path("proxy-ca.pem").exists():
        return True

    return False