"""Создание и проверка кодов регистрации Seaca.""" from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timedelta, timezone from sqlalchemy import func, select from sqlalchemy.orm import Session from app.models.mobile_device import MobileDevice from app.models.mobile_enrollment_code import MobileEnrollmentCode from app.models.mobile_settings import LOGIN_MODE_CODE_ONLY, LOGIN_MODE_PASSWORD, LOGIN_MODES from app.models.user import User from app.services.mobile_settings import get_effective_mobile_settings from app.services.mobile_tokens import ( generate_enrollment_code, hash_mobile_secret, revoke_refresh_tokens_for_device, store_refresh_token, ) from app.services.user_auth import authenticate_user @dataclass(frozen=True) class EnrollmentCodeSummary: id: int label: str code_prefix: str target_user_id: int | None target_username: str | None login_mode: str max_uses: int use_count: int expires_at: datetime revoked_at: datetime | None created_at: datetime is_active: bool @dataclass(frozen=True) class EnrollResult: access_token: str refresh_token: str device_id: int username: str role: str def _as_utc(dt: datetime) -> datetime: if dt.tzinfo is None: return dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc) def _code_is_usable(row: MobileEnrollmentCode, *, now: datetime) -> bool: if row.revoked_at is not None: return False if _as_utc(row.expires_at) <= now: return False if row.use_count >= row.max_uses: return False return True def list_enrollment_codes(db: Session) -> list[EnrollmentCodeSummary]: rows = db.scalars( select(MobileEnrollmentCode).order_by(MobileEnrollmentCode.created_at.desc()) ).all() usernames: dict[int, str] = {} user_ids = {r.target_user_id for r in rows if r.target_user_id} if user_ids: for user in db.scalars(select(User).where(User.id.in_(user_ids))).all(): usernames[user.id] = user.username now = datetime.now(timezone.utc) return [ EnrollmentCodeSummary( id=row.id, label=row.label or "", code_prefix=row.code_prefix, target_user_id=row.target_user_id, target_username=usernames.get(row.target_user_id) if row.target_user_id else None, login_mode=row.login_mode, max_uses=row.max_uses, use_count=row.use_count, expires_at=row.expires_at, revoked_at=row.revoked_at, created_at=row.created_at, is_active=_code_is_usable(row, now=now), ) for row in rows ] def create_enrollment_code( db: Session, *, created_by_user_id: int | None, label: str, target_user_id: int | None, login_mode: str, max_uses: int, expires_in_hours: int, ) -> tuple[EnrollmentCodeSummary, str]: if login_mode not in LOGIN_MODES: raise ValueError(f"login_mode must be one of: {sorted(LOGIN_MODES)}") if login_mode == LOGIN_MODE_CODE_ONLY and target_user_id is None: raise ValueError("code_only requires target_user_id") if max_uses < 1 or max_uses > 100: raise ValueError("max_uses must be between 1 and 100") if expires_in_hours < 1 or expires_in_hours > 24 * 30: raise ValueError("expires_in_hours must be between 1 and 720") if target_user_id is not None: user = db.get(User, target_user_id) if user is None or not user.is_active: raise ValueError("target user not found or inactive") plaintext, prefix, code_hash = generate_enrollment_code() expires_at = datetime.now(timezone.utc) + timedelta(hours=expires_in_hours) row = MobileEnrollmentCode( label=(label or "").strip(), code_hash=code_hash, code_prefix=prefix, created_by_user_id=created_by_user_id, target_user_id=target_user_id, login_mode=login_mode, max_uses=max_uses, use_count=0, expires_at=expires_at, ) db.add(row) db.flush() target_username = None if row.target_user_id: u = db.get(User, row.target_user_id) target_username = u.username if u else None summary = EnrollmentCodeSummary( id=row.id, label=row.label or "", code_prefix=row.code_prefix, target_user_id=row.target_user_id, target_username=target_username, login_mode=row.login_mode, max_uses=row.max_uses, use_count=row.use_count, expires_at=row.expires_at, revoked_at=row.revoked_at, created_at=row.created_at, is_active=True, ) return summary, plaintext def revoke_enrollment_code(db: Session, code_id: int) -> None: row = db.get(MobileEnrollmentCode, code_id) if row is None: raise ValueError("enrollment code not found") if row.revoked_at is None: row.revoked_at = datetime.now(timezone.utc) db.commit() def _count_active_devices(db: Session, user_id: int) -> int: return ( db.scalar( select(func.count()) .select_from(MobileDevice) .where(MobileDevice.user_id == user_id, MobileDevice.revoked_at.is_(None)) ) or 0 ) def enroll_device( db: Session, *, enrollment_code: str, username: str | None, password: str | None, device_uuid: str, display_name: str, platform: str, app_version: str | None, fcm_token: str | None, create_access_token, ) -> EnrollResult: from app.auth.jwt_auth import create_access_token as _create_access_token token_factory = create_access_token or _create_access_token mobile_cfg = get_effective_mobile_settings(db) if not mobile_cfg.devices_allowed: raise PermissionError("mobile enrollment is disabled") cleaned_uuid = (device_uuid or "").strip() if len(cleaned_uuid) < 8: raise ValueError("device_uuid is required") if mobile_cfg.min_app_version and app_version: if _version_lt(app_version, mobile_cfg.min_app_version): raise ValueError(f"app version {app_version} is below minimum {mobile_cfg.min_app_version}") now = datetime.now(timezone.utc) code_hash = hash_mobile_secret(enrollment_code.strip()) code_row = db.scalar( select(MobileEnrollmentCode).where(MobileEnrollmentCode.code_hash == code_hash) ) if code_row is None or not _code_is_usable(code_row, now=now): raise ValueError("invalid or expired enrollment code") if code_row.login_mode == LOGIN_MODE_PASSWORD: if not username or not password: raise ValueError("username and password required") user = authenticate_user(db, username, password) if user is None: raise ValueError("invalid username or password") else: user = db.get(User, code_row.target_user_id) if code_row.target_user_id else None if user is None or not user.is_active: raise ValueError("enrollment code user is invalid") if code_row.target_user_id is not None and user.id != code_row.target_user_id: raise ValueError("enrollment code is bound to another user") active_count = _count_active_devices(db, user.id) if active_count >= mobile_cfg.max_devices_per_user: raise ValueError("device limit reached for this user") existing = db.scalar(select(MobileDevice).where(MobileDevice.device_uuid == cleaned_uuid)) if existing is not None: if existing.revoked_at is None and existing.user_id != user.id: raise ValueError("device_uuid already registered to another user") if existing.revoked_at is None: revoke_refresh_tokens_for_device(db, existing.id) existing.revoked_at = now existing = None device = MobileDevice( user_id=user.id, device_uuid=cleaned_uuid, display_name=(display_name or "").strip() or "Android", platform=(platform or "android").strip() or "android", app_version=(app_version or "").strip() or None, fcm_token=(fcm_token or "").strip() or None, fcm_token_updated_at=now if fcm_token else None, enrollment_code_id=code_row.id, last_seen_at=now, ) db.add(device) code_row.use_count += 1 db.flush() refresh_plain = store_refresh_token(db, device_id=device.id) access_token = token_factory(user.username, user.role, device_id=device.id) db.commit() return EnrollResult( access_token=access_token, refresh_token=refresh_plain, device_id=device.id, username=user.username, role=user.role, ) def _version_lt(current: str, minimum: str) -> bool: def parse(v: str) -> tuple[int, ...]: parts: list[int] = [] for piece in v.strip().split("."): try: parts.append(int(piece)) except ValueError: parts.append(0) return tuple(parts) return parse(current) < parse(minimum)