Files
security-alert-center/backend/app/services/mobile_tokens.py
T

69 lines
2.0 KiB
Python

"""Хеширование enrollment/refresh токенов и генерация."""
from __future__ import annotations
import hashlib
import secrets
from datetime import datetime, timedelta, timezone
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.config import get_settings
from app.models.mobile_refresh_token import MobileRefreshToken
def hash_mobile_secret(raw: str) -> str:
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
def generate_enrollment_code() -> tuple[str, str, str]:
"""(plaintext, prefix, hash)"""
raw = f"sacmob_{secrets.token_urlsafe(18)}"
prefix = raw[:12]
return raw, prefix, hash_mobile_secret(raw)
def generate_refresh_token() -> tuple[str, str]:
"""(plaintext, hash)"""
raw = secrets.token_urlsafe(48)
return raw, hash_mobile_secret(raw)
def store_refresh_token(db: Session, *, device_id: int) -> str:
settings = get_settings()
days = max(1, int(settings.sac_mobile_refresh_expire_days))
plaintext, token_hash = generate_refresh_token()
expires_at = datetime.now(timezone.utc) + timedelta(days=days)
row = MobileRefreshToken(
device_id=device_id,
token_hash=token_hash,
expires_at=expires_at,
)
db.add(row)
return plaintext
def revoke_refresh_tokens_for_device(db: Session, device_id: int) -> None:
now = datetime.now(timezone.utc)
rows = db.scalars(
select(MobileRefreshToken).where(
MobileRefreshToken.device_id == device_id,
MobileRefreshToken.revoked_at.is_(None),
)
).all()
for row in rows:
row.revoked_at = now
def find_valid_refresh_token(db: Session, raw_token: str) -> MobileRefreshToken | None:
token_hash = hash_mobile_secret(raw_token)
now = datetime.now(timezone.utc)
return db.scalar(
select(MobileRefreshToken).where(
MobileRefreshToken.token_hash == token_hash,
MobileRefreshToken.revoked_at.is_(None),
MobileRefreshToken.expires_at > now,
)
)