Add configurable TLS handling for GigaChat

serv/.env.example

@@ -8,6 +8,8 @@ NEW_QWEN_GIGACHAT_AUTH_KEY=
 NEW_QWEN_GIGACHAT_SCOPE=GIGACHAT_API_PERS
 NEW_QWEN_GIGACHAT_API_BASE_URL=https://gigachat.devices.sberbank.ru/api/v1
 NEW_QWEN_GIGACHAT_OAUTH_URL=https://ngw.devices.sberbank.ru:9443/api/v2/oauth
+NEW_QWEN_GIGACHAT_CA_BUNDLE=
+NEW_QWEN_GIGACHAT_INSECURE_SKIP_VERIFY=
 NEW_QWEN_YANDEXGPT_MODEL=yandexgpt
 NEW_QWEN_WORKSPACE_ROOT=/home/mirivlad/git
 NEW_QWEN_SESSION_DIR=/home/mirivlad/git/new-qwen/.new-qwen/sessions

serv/config.py

@@ -28,6 +28,8 @@ class ServerConfig:
     gigachat_scope: str
     gigachat_api_base_url: str
     gigachat_oauth_url: str
+    gigachat_ca_bundle: Path | None
+    gigachat_insecure_skip_verify: bool
     yandexgpt_model: str
     workspace_root: Path
     session_dir: Path
@@ -84,6 +86,15 @@ class ServerConfig:
                 "NEW_QWEN_GIGACHAT_OAUTH_URL",
                 "https://ngw.devices.sberbank.ru:9443/api/v2/oauth",
             ).strip(),
+            gigachat_ca_bundle=(
+                Path(os.environ["NEW_QWEN_GIGACHAT_CA_BUNDLE"]).expanduser().resolve()
+                if os.environ.get("NEW_QWEN_GIGACHAT_CA_BUNDLE", "").strip()
+                else None
+            ),
+            gigachat_insecure_skip_verify=os.environ.get(
+                "NEW_QWEN_GIGACHAT_INSECURE_SKIP_VERIFY",
+                "",
+            ).strip().lower() in {"1", "true", "yes", "on"},
             yandexgpt_model=os.environ.get("NEW_QWEN_YANDEXGPT_MODEL", "yandexgpt").strip(),
             workspace_root=workspace_root.resolve(),
             session_dir=session_dir.resolve(),

serv/gigachat.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import ssl
 import time
 import uuid
 from typing import Any
@@ -22,6 +23,16 @@ class GigaChatAuthManager:
     def is_configured(self) -> bool:
         return bool(self.config.gigachat_auth_key)
 
+    def build_ssl_context(self) -> ssl.SSLContext:
+        if self.config.gigachat_insecure_skip_verify:
+            return ssl._create_unverified_context()
+        if self.config.gigachat_ca_bundle:
+            return ssl.create_default_context(cafile=str(self.config.gigachat_ca_bundle))
+        return ssl.create_default_context()
+
+    def open(self, req: request.Request | str, timeout: int):
+        return request.urlopen(req, timeout=timeout, context=self.build_ssl_context())
+
     def _authorization_header(self) -> str:
         raw = self.config.gigachat_auth_key.strip()
         if not raw:
@@ -58,7 +69,7 @@ class GigaChatAuthManager:
             method="POST",
         )
         try:
-            with request.urlopen(req, timeout=60) as response:
+            with self.open(req, timeout=60) as response:
                 payload = json.loads(response.read().decode("utf-8"))
         except error.HTTPError as exc:
             body = exc.read().decode("utf-8", errors="replace")

serv/model_router.py

@@ -315,7 +315,7 @@ class GigaChatModelProvider(BaseModelProvider):
             method="POST",
         )
         try:
-            with request.urlopen(req, timeout=180) as response:
+            with self.auth.open(req, timeout=180) as response:
                 raw = json.loads(response.read().decode("utf-8"))
         except error.HTTPError as exc:
             body = exc.read().decode("utf-8", errors="replace")