feat(auth): users module with bcrypt hashing + CRUD
This commit is contained in:
@@ -0,0 +1 @@
|
||||
"""Auth module — users, sessions, permissions, routes, admin, rate_limit."""
|
||||
@@ -0,0 +1,79 @@
|
||||
"""User CRUD + bcrypt password hashing."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from passlib.hash import bcrypt
|
||||
from sqlalchemy import select
|
||||
|
||||
from cyclone.db import User
|
||||
|
||||
|
||||
def hash_password(plaintext: str) -> str:
|
||||
return bcrypt.hash(plaintext)
|
||||
|
||||
|
||||
def verify_password(plaintext: str, hashed: str) -> bool:
|
||||
try:
|
||||
return bcrypt.verify(plaintext, hashed)
|
||||
except (ValueError, TypeError):
|
||||
return False
|
||||
|
||||
|
||||
def create(db, *, username: str, password: str, role: str) -> User:
|
||||
user = User(
|
||||
username=username,
|
||||
password_hash=hash_password(password),
|
||||
role=role,
|
||||
created_at=datetime.now(timezone.utc),
|
||||
)
|
||||
db.add(user)
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
return user
|
||||
|
||||
|
||||
def get_by_username(db, username: str) -> User | None:
|
||||
return db.execute(
|
||||
select(User).where(User.username == username)
|
||||
).scalar_one_or_none()
|
||||
|
||||
|
||||
def get(db, user_id: int) -> User | None:
|
||||
return db.get(User, user_id)
|
||||
|
||||
|
||||
def disable(db, user_id: int) -> None:
|
||||
user = db.get(User, user_id)
|
||||
if user is None:
|
||||
return
|
||||
user.disabled_at = datetime.now(timezone.utc)
|
||||
db.commit()
|
||||
|
||||
|
||||
def update_role(db, user_id: int, role: str) -> None:
|
||||
user = db.get(User, user_id)
|
||||
if user is None:
|
||||
return
|
||||
user.role = role
|
||||
db.commit()
|
||||
|
||||
|
||||
def update_password(db, user_id: int, new_password: str) -> None:
|
||||
user = db.get(User, user_id)
|
||||
if user is None:
|
||||
return
|
||||
user.password_hash = hash_password(new_password)
|
||||
db.commit()
|
||||
|
||||
|
||||
def to_public(user: User) -> dict[str, Any]:
|
||||
return {
|
||||
"id": user.id,
|
||||
"username": user.username,
|
||||
"role": user.role,
|
||||
"createdAt": user.created_at.isoformat() if user.created_at else None,
|
||||
"disabledAt": user.disabled_at.isoformat() if user.disabled_at else None,
|
||||
}
|
||||
Reference in New Issue
Block a user