55 lines
1.1 KiB
Python
55 lines
1.1 KiB
Python
import secrets
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from domains.users.models import User
|
|
from domains.users.repo import (
|
|
get_user_by_id,
|
|
get_user_by_username,
|
|
create_user,
|
|
)
|
|
|
|
from core.security import hash_password
|
|
from core.config import settings
|
|
|
|
|
|
def generate_user_secret():
|
|
return secrets.token_urlsafe(settings.SECRET_PASS_LENGTH)
|
|
|
|
async def admin_create_user(
|
|
db: AsyncSession,
|
|
username: str,
|
|
role: str
|
|
):
|
|
|
|
existing = await get_user_by_username(db, username)
|
|
|
|
if existing:
|
|
raise ValueError("Username already exists")
|
|
|
|
secret = generate_user_secret()
|
|
|
|
user = User(
|
|
username=username,
|
|
role=role,
|
|
secret_hash=hash_password(secret)
|
|
)
|
|
|
|
await create_user(db, user)
|
|
|
|
return user, secret
|
|
|
|
|
|
async def admin_reset_user_secret(
|
|
db: AsyncSession,
|
|
user_id
|
|
):
|
|
|
|
user = await get_user_by_id(db, user_id)
|
|
|
|
if not user:
|
|
return None
|
|
new_secret = generate_user_secret()
|
|
user.secret_hash = hash_password(new_secret)
|
|
await db.commit()
|
|
return new_secret |