security-alert-center/backend/app/services/mobile_enrollment.py

"""Создание и проверка кодов регистрации 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)