112 lines
3.4 KiB
Python
112 lines
3.4 KiB
Python
"""HMAC request signing and local JWT token persistence."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import binascii
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
import time
|
||
import uuid
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from dotenv import set_key
|
||
|
||
from .errors import ApiError
|
||
|
||
|
||
def jwt_expiration(token: str) -> int:
|
||
"""Read a JWT exp claim; this does not replace server-side verification."""
|
||
try:
|
||
encoded_payload = token.split(".")[1]
|
||
padding = "=" * (-len(encoded_payload) % 4)
|
||
payload = json.loads(
|
||
base64.urlsafe_b64decode(encoded_payload + padding).decode("utf-8")
|
||
)
|
||
return int(payload["exp"])
|
||
except (
|
||
IndexError,
|
||
KeyError,
|
||
TypeError,
|
||
ValueError,
|
||
UnicodeDecodeError,
|
||
binascii.Error,
|
||
) as exc:
|
||
raise ApiError("API geçerli bir JWT token döndürmedi.") from exc
|
||
|
||
|
||
def token_is_valid(token: str | None, skew_seconds: int) -> bool:
|
||
if not token:
|
||
return False
|
||
try:
|
||
return jwt_expiration(token) > int(time.time()) + skew_seconds
|
||
except ApiError:
|
||
return False
|
||
|
||
|
||
def create_signed_headers(
|
||
*,
|
||
method: str,
|
||
path_with_query: str,
|
||
body: str,
|
||
secret: str,
|
||
accept_language: str,
|
||
idempotency_key: str,
|
||
) -> dict[str, str]:
|
||
timestamp = str(int(time.time() * 1000))
|
||
data_to_sign = f"{method.upper()}|{path_with_query}|{timestamp}|{body}"
|
||
signature = hmac.new(
|
||
secret.encode("utf-8"),
|
||
data_to_sign.encode("utf-8"),
|
||
hashlib.sha256,
|
||
).hexdigest()
|
||
return {
|
||
"Accept": "application/json",
|
||
"Accept-Language": accept_language,
|
||
"Content-Type": "application/json",
|
||
"X-Signature": signature,
|
||
"X-Timestamp": timestamp,
|
||
"X-Nonce": str(uuid.uuid4()),
|
||
"X-Idempotency-Key": idempotency_key,
|
||
}
|
||
|
||
|
||
class TokenStore:
|
||
def __init__(self, env_file: Path) -> None:
|
||
self.env_file = env_file
|
||
|
||
@property
|
||
def access_token(self) -> str | None:
|
||
return os.getenv("IKLIM_ACCESS_TOKEN", "").strip() or None
|
||
|
||
@property
|
||
def refresh_token(self) -> str | None:
|
||
return os.getenv("IKLIM_REFRESH_TOKEN", "").strip() or None
|
||
|
||
def save(self, response_data: dict[str, Any]) -> None:
|
||
access_token = response_data.get("accessToken")
|
||
refresh_token = response_data.get("refreshToken")
|
||
if not isinstance(access_token, str) or not access_token:
|
||
raise ApiError("Kimlik doğrulama yanıtında accessToken bulunamadı.")
|
||
if not isinstance(refresh_token, str) or not refresh_token:
|
||
raise ApiError("Kimlik doğrulama yanıtında refreshToken bulunamadı.")
|
||
|
||
values = {
|
||
"IKLIM_ACCESS_TOKEN": access_token,
|
||
"IKLIM_REFRESH_TOKEN": refresh_token,
|
||
"IKLIM_ACCESS_TOKEN_EXPIRES_AT": str(jwt_expiration(access_token)),
|
||
"IKLIM_REFRESH_TOKEN_EXPIRES_AT": str(jwt_expiration(refresh_token)),
|
||
"IKLIM_TOKENS_UPDATED_AT": str(int(time.time())),
|
||
}
|
||
for key, value in values.items():
|
||
set_key(self.env_file, key, value, quote_mode="never")
|
||
os.environ[key] = value
|
||
|
||
def set_webhook_initialized(self, initialized: bool) -> None:
|
||
value = "true" if initialized else "false"
|
||
set_key(self.env_file, "IKLIM_WEBHOOK_INITIALIZED", value, quote_mode="never")
|
||
os.environ["IKLIM_WEBHOOK_INITIALIZED"] = value
|