Class DeviceCode

enterprise/storage/device_code.py:20–131

Device code for OAuth 2.0 Device Flow. This stores the device codes issued during the device authorization flow, along with their status and associated user information once authorized.

class DeviceCode(Base):
    """Device code for OAuth 2.0 Device Flow.

    This stores the device codes issued during the device authorization flow,
    along with their status and associated user information once authorized.
    """

    __tablename__ = 'device_codes'

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    device_code: Mapped[str] = mapped_column(
        String(128), unique=True, nullable=False, index=True
    )
    user_code: Mapped[str] = mapped_column(
        String(16), unique=True, nullable=False, index=True
    )
    status: Mapped[str] = mapped_column(
        String(32), nullable=False, default=DeviceCodeStatus.PENDING.value
    )

    # Keycloak user ID who authorized the device (set during verification)
    keycloak_user_id: Mapped[str | None] = mapped_column(String(255), nullable=True)

    # Timestamps
    expires_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False
    )
    authorized_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )

    # Rate limiting fields for RFC 8628 section 3.5 compliance
    last_poll_time: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    current_interval: Mapped[int] = mapped_column(nullable=False, default=5)

    def __repr__(self) -> str:
        return f"<DeviceCode(user_code='{self.user_code}', status='{self.status}')>"

    def is_expired(self) -> bool:
        """Check if the device code has expired."""
        now = datetime.now(timezone.utc)
        # Handle timezone-naive datetime from database by assuming it's UTC
        expires_at = self.expires_at
        if expires_at.tzinfo is None:
            expires_at = expires_at.replace(tzinfo=timezone.utc)
        return now > expires_at

    def is_pending(self) -> bool:
        """Check if the device code is still pending authorization."""
        return self.status == DeviceCodeStatus.PENDING.value and not self.is_expired()

    def is_authorized(self) -> bool:
        """Check if the device code has been authorized."""
        return self.status == DeviceCodeStatus.AUTHORIZED.value

    def authorize(self, user_id: str) -> None: