Function create_user_without_org

apps/api/src/services/users/users.py:263–348

(
    request: Request,
    db_session: AsyncSession,
    current_user: PublicUser | AnonymousUser,
    user_object: UserCreate,
    is_oauth: bool = False,
    signup_provider: str = "email",
)

async def create_user_without_org(
    request: Request,
    db_session: AsyncSession,
    current_user: PublicUser | AnonymousUser,
    user_object: UserCreate,
    is_oauth: bool = False,
    signup_provider: str = "email",
):
    # Validate password complexity (skip for OAuth users who have empty passwords)
    if user_object.password and not is_oauth:
        validation_result = validate_password_complexity(user_object.password)
        if not validation_result.is_valid:
            raise HTTPException(
                status_code=400,
                detail={
                    "code": "WEAK_PASSWORD",
                    "message": "Password does not meet security requirements",
                    "errors": validation_result.errors,
                    "requirements": validation_result.requirements,
                },
            )

    user = User.model_validate(user_object)

    # RBAC check
    await rbac_check(request, current_user, "create", "user_x", db_session)

    # Complete the user object
    user.user_uuid = f"user_{uuid4()}"
    user.password = security_hash_password(user_object.password) if user_object.password else ""

    # OAuth users and OSS mode get auto-verified email
    if is_oauth or get_deployment_mode() != 'saas':
        user.email_verified = True
        user.email_verified_at = datetime.now(timezone.utc).isoformat()
        user.signup_method = signup_provider if is_oauth else "email"
    else:
        user.email_verified = False
        user.email_verified_at = None
        user.signup_method = "email"

    user.creation_date = str(datetime.now())
    user.update_date = str(datetime.now())

    # Verifications

    # SECURITY: single generic error for both email and username conflicts to
    # prevent account enumeration via this org-less signup endpoint.
    conflict = (await db_session.execute(
        select(User).where(
            (User.username == user.username) | (User.email == user.email)
        )
    )).scalars().first()

    if conflict:
        raise HTTPException(
            status_code=400,
            detail="Email or username is already in use",