249 lines
6.7 KiB
Python
249 lines
6.7 KiB
Python
"""User model for authentication"""
|
|||
|
|
import os
|
||
|
|
import yaml
|
||
|
|
from flask_login import UserMixin
|
||
|
|
import bcrypt
|
||
|
|
|
||
|
|
|
||
|
|
class User(UserMixin):
|
||
|
|
"""User model for product owners and administrators
|
||
|
|
|
||
|
|
Attributes:
|
||
|
|
user_id: Unique user identifier
|
||
|
|
username: Username for login
|
||
|
|
email: User email address
|
||
|
|
password_hash: Bcrypt hashed password
|
||
|
|
role: User role ('product_owner' or 'administrator')
|
||
|
|
product_ids: List of product IDs (for product_owner role)
|
||
|
|
is_active: Whether user account is active
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self, user_id, username, email, password_hash, role, product_ids=None, is_active=True):
|
||
|
|
self.user_id = user_id
|
||
|
|
self.username = username
|
||
|
|
self.email = email
|
||
|
|
self.password_hash = password_hash
|
||
|
|
self.role = role
|
||
|
|
self.product_ids = product_ids or []
|
||
|
|
self.is_active = is_active
|
||
|
|
|
||
|
|
def get_id(self):
|
||
|
|
"""Get user ID for Flask-Login"""
|
||
|
|
return self.user_id
|
||
|
|
|
||
|
|
@property
|
||
|
|
def is_authenticated(self):
|
||
|
|
"""Check if user is authenticated"""
|
||
|
|
return True
|
||
|
|
|
||
|
|
@property
|
||
|
|
def is_anonymous(self):
|
||
|
|
"""Check if user is anonymous"""
|
||
|
|
return False
|
||
|
|
|
||
|
|
def check_password(self, password):
|
||
|
|
"""Verify password against stored hash
|
||
|
|
|
||
|
|
Args:
|
||
|
|
password: Plain text password to verify
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
bool: True if password matches, False otherwise
|
||
|
|
"""
|
||
|
|
return bcrypt.checkpw(password.encode('utf-8'), self.password_hash.encode('utf-8'))
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def hash_password(password):
|
||
|
|
"""Hash password using bcrypt
|
||
|
|
|
||
|
|
Args:
|
||
|
|
password: Plain text password
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
str: Hashed password
|
||
|
|
"""
|
||
|
|
salt = bcrypt.gensalt()
|
||
|
|
return bcrypt.hashpw(password.encode('utf-8'), salt).decode('utf-8')
|
||
|
|
|
||
|
|
def to_dict(self):
|
||
|
|
"""Convert user to dictionary for storage
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
dict: User data
|
||
|
|
"""
|
||
|
|
return {
|
||
|
|
'user_id': self.user_id,
|
||
|
|
'username': self.username,
|
||
|
|
'email': self.email,
|
||
|
|
'password_hash': self.password_hash,
|
||
|
|
'role': self.role,
|
||
|
|
'product_ids': self.product_ids,
|
||
|
|
'is_active': self.is_active
|
||
|
|
}
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def from_dict(cls, data):
|
||
|
|
"""Create user from dictionary
|
||
|
|
|
||
|
|
Args:
|
||
|
|
data: Dictionary with user data
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
User: User instance
|
||
|
|
"""
|
||
|
|
return cls(
|
||
|
|
user_id=data['user_id'],
|
||
|
|
username=data['username'],
|
||
|
|
email=data['email'],
|
||
|
|
password_hash=data['password_hash'],
|
||
|
|
role=data['role'],
|
||
|
|
product_ids=data.get('product_ids', []),
|
||
|
|
is_active=data.get('is_active', True)
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _get_users_file():
|
||
|
|
"""Get path to users YAML file
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
str: Path to users.yaml
|
||
|
|
"""
|
||
|
|
from flask import current_app
|
||
|
|
return os.path.join(current_app.config['DATA_DIR'], 'users.yaml')
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _load_all_users():
|
||
|
|
"""Load all users from storage
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
dict: Dictionary of user_id -> user_data
|
||
|
|
"""
|
||
|
|
users_file = User._get_users_file()
|
||
|
|
|
||
|
|
if not os.path.exists(users_file):
|
||
|
|
return {}
|
||
|
|
|
||
|
|
with open(users_file, 'r') as f:
|
||
|
|
data = yaml.safe_load(f) or {}
|
||
|
|
return data.get('users', {})
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _save_all_users(users_dict):
|
||
|
|
"""Save all users to storage
|
||
|
|
|
||
|
|
Args:
|
||
|
|
users_dict: Dictionary of user_id -> user_data
|
||
|
|
"""
|
||
|
|
users_file = User._get_users_file()
|
||
|
|
os.makedirs(os.path.dirname(users_file), exist_ok=True)
|
||
|
|
|
||
|
|
with open(users_file, 'w') as f:
|
||
|
|
yaml.dump({'users': users_dict}, f, default_flow_style=False)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def get_by_id(cls, user_id):
|
||
|
|
"""Load user by ID
|
||
|
|
|
||
|
|
Args:
|
||
|
|
user_id: User ID to load
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
User or None: User instance if found, None otherwise
|
||
|
|
"""
|
||
|
|
users = cls._load_all_users()
|
||
|
|
user_data = users.get(user_id)
|
||
|
|
|
||
|
|
if user_data:
|
||
|
|
return cls.from_dict(user_data)
|
||
|
|
return None
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def get_by_username(cls, username):
|
||
|
|
"""Load user by username
|
||
|
|
|
||
|
|
Args:
|
||
|
|
username: Username to search for
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
User or None: User instance if found, None otherwise
|
||
|
|
"""
|
||
|
|
users = cls._load_all_users()
|
||
|
|
|
||
|
|
for user_data in users.values():
|
||
|
|
if user_data['username'] == username:
|
||
|
|
return cls.from_dict(user_data)
|
||
|
|
return None
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def get_all(cls):
|
||
|
|
"""Get all users
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
list: List of User instances
|
||
|
|
"""
|
||
|
|
users = cls._load_all_users()
|
||
|
|
return [cls.from_dict(data) for data in users.values()]
|
||
|
|
|
||
|
|
def save(self):
|
||
|
|
"""Save user to storage"""
|
||
|
|
users = self._load_all_users()
|
||
|
|
users[self.user_id] = self.to_dict()
|
||
|
|
self._save_all_users(users)
|
||
|
|
|
||
|
|
def delete(self):
|
||
|
|
"""Delete user from storage"""
|
||
|
|
users = self._load_all_users()
|
||
|
|
if self.user_id in users:
|
||
|
|
del users[self.user_id]
|
||
|
|
self._save_all_users(users)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def create(cls, username, email, password, role, product_ids=None):
|
||
|
|
"""Create new user
|
||
|
|
|
||
|
|
Args:
|
||
|
|
username: Username for login
|
||
|
|
email: User email
|
||
|
|
password: Plain text password
|
||
|
|
role: User role ('product_owner' or 'administrator')
|
||
|
|
product_ids: List of product IDs (for product_owner)
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
User: Created user instance
|
||
|
|
|
||
|
|
Raises:
|
||
|
|
ValueError: If username already exists or role is invalid
|
||
|
|
"""
|
||
|
|
# Validate role
|
||
|
|
if role not in ['product_owner', 'administrator']:
|
||
|
|
raise ValueError(f"Invalid role: {role}")
|
||
|
|
|
||
|
|
# Check if username exists
|
||
|
|
if cls.get_by_username(username):
|
||
|
|
raise ValueError(f"Username already exists: {username}")
|
||
|
|
|
||
|
|
# Generate user ID
|
||
|
|
users = cls._load_all_users()
|
||
|
|
if users:
|
||
|
|
max_id = max([int(uid.replace('usr_', '')) for uid in users.keys()])
|
||
|
|
user_id = f"usr_{max_id + 1:04d}"
|
||
|
|
else:
|
||
|
|
user_id = "usr_0001"
|
||
|
|
|
||
|
|
# Hash password
|
||
|
|
password_hash = cls.hash_password(password)
|
||
|
|
|
||
|
|
# Create user
|
||
|
|
user = cls(
|
||
|
|
user_id=user_id,
|
||
|
|
username=username,
|
||
|
|
email=email,
|
||
|
|
password_hash=password_hash,
|
||
|
|
role=role,
|
||
|
|
product_ids=product_ids or [],
|
||
|
|
is_active=True
|
||
|
|
)
|
||
|
|
|
||
|
|
user.save()
|
||
|
|
return user
|