88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
import json
|
||
import os
|
||
import ssl
|
||
import urllib.error
|
||
import urllib.request
|
||
import urllib.parse
|
||
from typing import Optional
|
||
|
||
|
||
DEFAULT_USERNAME = "+79214400842"
|
||
DEFAULT_PASSWORD = "stepan"
|
||
DEFAULT_GRANT_TYPE = "password"
|
||
|
||
|
||
def _build_ssl_context(url: str) -> ssl.SSLContext | None:
|
||
"""
|
||
На dev-стендах сертификат иногда не проходит системную валидацию на Windows.
|
||
По умолчанию для *.dev.dipal.ru отключаем verify, но это можно переопределить env SSL_VERIFY=1.
|
||
"""
|
||
try:
|
||
parsed = urllib.parse.urlparse(url)
|
||
except Exception:
|
||
return None
|
||
|
||
if (os.getenv("SSL_VERIFY") or "").lower() in {"1", "true", "yes"}:
|
||
return None
|
||
|
||
if (parsed.scheme or "").lower() != "https":
|
||
return None
|
||
|
||
host = (parsed.hostname or "").lower()
|
||
if host.endswith(".dev.dipal.ru") or host == "dev.dipal.ru":
|
||
return ssl._create_unverified_context() # noqa: SLF001
|
||
|
||
return None
|
||
|
||
|
||
def get_access_token(
|
||
auth_url: Optional[str] = None,
|
||
*,
|
||
username: str = DEFAULT_USERNAME,
|
||
password: str = DEFAULT_PASSWORD,
|
||
grant_type: str = DEFAULT_GRANT_TYPE,
|
||
timeout_s: int = 30,
|
||
) -> str:
|
||
"""
|
||
POST авторизация, возвращает access_token.
|
||
|
||
Можно переопределить AUTH_URL через env.
|
||
"""
|
||
auth_url = auth_url or os.getenv("AUTH_URL") or "https://auth.dev.dipal.ru/api/v1/auth/login"
|
||
if not auth_url:
|
||
raise ValueError("Не задан AUTH_URL (передайте auth_url или установите env AUTH_URL)")
|
||
|
||
# Поддерживаем env-переопределение кредов, чтобы не хардкодить их в тестах.
|
||
username = os.getenv("AUTH_USERNAME", username)
|
||
password = os.getenv("AUTH_PASSWORD", password)
|
||
grant_type = os.getenv("AUTH_GRANT_TYPE", grant_type)
|
||
|
||
payload = {"username": username, "password": password, "grant_type": grant_type}
|
||
req = urllib.request.Request(
|
||
auth_url,
|
||
data=json.dumps(payload).encode("utf-8"),
|
||
headers={"content-type": "application/json"},
|
||
method="POST",
|
||
)
|
||
try:
|
||
ctx = _build_ssl_context(auth_url)
|
||
with urllib.request.urlopen(req, timeout=timeout_s, context=ctx) as resp:
|
||
raw = resp.read().decode("utf-8")
|
||
except urllib.error.HTTPError as e:
|
||
body = e.read().decode("utf-8", errors="replace")
|
||
raise RuntimeError(f"Auth HTTP {e.code}: {body}") from e
|
||
|
||
data = json.loads(raw) if raw else {}
|
||
if "data" in data and isinstance(data["data"], dict):
|
||
token = data["data"].get("access_token")
|
||
else:
|
||
token = data.get("access_token")
|
||
if not token:
|
||
raise ValueError(f"В ответе нет access_token. Ключи ответа: {list(data.keys())}")
|
||
return token
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print(get_access_token())
|
||
|