123 lines
4.4 KiB
Python
123 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Client HTTP con sessione cookie (Loogle Casa, Irrigazione)."""
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
from typing import Any, Optional
|
|
from urllib.parse import urljoin
|
|
|
|
import httpx
|
|
|
|
LOGGER = logging.getLogger("loogle_mcp.session_client")
|
|
|
|
_sessions: dict[str, tuple[str, float]] = {}
|
|
_sessions_lock = threading.Lock()
|
|
SESSION_TTL = 3600 * 12
|
|
|
|
|
|
class SessionApiClient:
|
|
"""Login cookie-based con cache per utente MCP."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
service: str,
|
|
base_url: str,
|
|
login_path: str = "/api/login",
|
|
verify_ssl: bool = True,
|
|
) -> None:
|
|
self.service = service
|
|
self.base_url = base_url.rstrip("/")
|
|
self.login_path = login_path
|
|
self.verify_ssl = verify_ssl
|
|
|
|
def _password_for_user(self, username: str) -> Optional[str]:
|
|
import os
|
|
user = username.lower()
|
|
env_key = f"{self.service}_PASSWORD_{user.upper()}"
|
|
pwd = os.environ.get(env_key, "").strip()
|
|
if pwd:
|
|
return pwd
|
|
fallback = os.environ.get(f"{self.service}_PASSWORD", "").strip()
|
|
if fallback:
|
|
return fallback
|
|
return user
|
|
|
|
def _cache_key(self, username: str) -> str:
|
|
return f"{self.service}:{username.lower()}"
|
|
|
|
def _get_cached_cookie(self, username: str) -> Optional[str]:
|
|
key = self._cache_key(username)
|
|
with _sessions_lock:
|
|
row = _sessions.get(key)
|
|
if not row:
|
|
return None
|
|
cookie, expires = row
|
|
if time.time() > expires:
|
|
_sessions.pop(key, None)
|
|
return None
|
|
return cookie
|
|
|
|
def _store_cookie(self, username: str, cookie: str) -> None:
|
|
key = self._cache_key(username)
|
|
with _sessions_lock:
|
|
_sessions[key] = (cookie, time.time() + SESSION_TTL)
|
|
|
|
def login(self, username: str) -> str:
|
|
cached = self._get_cached_cookie(username)
|
|
if cached:
|
|
return cached
|
|
service_user = username
|
|
if hasattr(self, "map_username"):
|
|
service_user = self.map_username(username) # type: ignore[attr-defined]
|
|
password = self._password_for_user(username)
|
|
if not password:
|
|
raise RuntimeError(
|
|
f"Password {self.service} non configurata per {username}. "
|
|
f"Imposta {self.service}_PASSWORD_{username.upper()} in .env"
|
|
)
|
|
url = urljoin(self.base_url + "/", self.login_path.lstrip("/"))
|
|
with httpx.Client(timeout=30.0, verify=self.verify_ssl) as client:
|
|
resp = client.post(
|
|
url, json={"username": service_user, "password": password},
|
|
)
|
|
if resp.status_code != 200:
|
|
raise RuntimeError(
|
|
f"Login {self.service} fallito per {username}: HTTP {resp.status_code}"
|
|
)
|
|
cookie = resp.cookies.get("session")
|
|
if not cookie:
|
|
raise RuntimeError(f"Login {self.service}: cookie session mancante")
|
|
self._store_cookie(username, cookie)
|
|
return cookie
|
|
|
|
def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
username: str,
|
|
params: Optional[dict] = None,
|
|
json_body: Optional[dict] = None,
|
|
) -> Any:
|
|
cookie = self.login(username)
|
|
url = path if path.startswith("http") else urljoin(self.base_url + "/", path.lstrip("/"))
|
|
headers = {"Cookie": f"session={cookie}"}
|
|
with httpx.Client(timeout=60.0, verify=self.verify_ssl) as client:
|
|
resp = client.request(method, url, headers=headers, params=params, json=json_body)
|
|
if resp.status_code == 401:
|
|
with _sessions_lock:
|
|
_sessions.pop(self._cache_key(username), None)
|
|
cookie = self.login(username)
|
|
headers = {"Cookie": f"session={cookie}"}
|
|
resp = client.request(method, url, headers=headers, params=params, json=json_body)
|
|
if resp.status_code >= 400:
|
|
raise RuntimeError(f"{self.service} {method} {path}: HTTP {resp.status_code} {resp.text[:200]}")
|
|
if resp.headers.get("content-type", "").startswith("application/json"):
|
|
return resp.json()
|
|
return resp.text
|
|
|
|
def get(self, path: str, *, username: str, params: Optional[dict] = None) -> Any:
|
|
return self.request("GET", path, username=username, params=params)
|