PTah
563b836acc
Adds mobile device registration by admin codes, refresh tokens, push channel in notification policy, and Settings section for managing Seaca clients. Co-authored-by: Cursor <cursoragent@cursor.com>
69 lines
2.0 KiB
Python
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,
|
|
)
|
|
)
|