SkyForces
99 lines
2.9 KiB
Python
99 lines
2.9 KiB
Python
import hashlib
|
|
import secrets
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from .database import SiteDatabase
|
|
from .crypto import TokenCipher
|
|
|
|
|
|
def hash_token(token: str) -> bytes:
|
|
return hashlib.sha256(token.encode("utf-8")).digest()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BindingLink:
|
|
url: str
|
|
expires_at: datetime
|
|
|
|
|
|
class SiteBindingRepository:
|
|
"""Доступ сайта только к функциям выпуска токенов."""
|
|
|
|
def __init__(self, database: SiteDatabase) -> None:
|
|
self.database = database
|
|
|
|
def issue(
|
|
self,
|
|
member_id: str,
|
|
domain: str,
|
|
bitrix_user_id: int,
|
|
token_hash: bytes,
|
|
token_expires_at: datetime,
|
|
access_token: bytes,
|
|
refresh_token: bytes,
|
|
oauth_expires_at: datetime,
|
|
) -> None:
|
|
"""Сохраняет в БД информацию о токене, выданном пользователю."""
|
|
query = """
|
|
SELECT *
|
|
FROM binding.issue_v1(%s, %s, %s, %s, %s, %s, %s, %s) \
|
|
"""
|
|
with self.database.transaction() as connection:
|
|
connection.execute(
|
|
query,
|
|
(
|
|
member_id,
|
|
domain,
|
|
bitrix_user_id,
|
|
token_hash,
|
|
token_expires_at,
|
|
access_token,
|
|
refresh_token,
|
|
oauth_expires_at,
|
|
),
|
|
).fetchone()
|
|
|
|
|
|
class BindingService:
|
|
def __init__(
|
|
self,
|
|
repository: SiteBindingRepository,
|
|
cipher: TokenCipher,
|
|
bot_username: str,
|
|
ttl_seconds: int,
|
|
) -> None:
|
|
self.repository = repository
|
|
self.cipher = cipher
|
|
self.bot_username = bot_username
|
|
self.ttl_seconds = ttl_seconds
|
|
|
|
def issue(
|
|
self,
|
|
member_id: str,
|
|
domain: str,
|
|
bitrix_user_id: int,
|
|
access_token: str,
|
|
refresh_token: str,
|
|
auth_expires_seconds: int,
|
|
) -> BindingLink:
|
|
"""Выдает ссылку для привязки аккаунта."""
|
|
token = secrets.token_urlsafe(32)
|
|
now = datetime.now(UTC)
|
|
token_expires_at = now + timedelta(seconds=self.ttl_seconds)
|
|
oauth_expires_at = now + timedelta(seconds=auth_expires_seconds)
|
|
self.repository.issue(
|
|
member_id,
|
|
domain,
|
|
bitrix_user_id,
|
|
hash_token(token),
|
|
token_expires_at,
|
|
self.cipher.encrypt(access_token),
|
|
self.cipher.encrypt(refresh_token),
|
|
oauth_expires_at,
|
|
)
|
|
return BindingLink(
|
|
url=f"https://t.me/{self.bot_username}?start=bind_{token}",
|
|
expires_at=token_expires_at,
|
|
)
|