mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-09 05:24:49 +09:00
Juan Denis
b7fab1152b
Introduce team invitation support and a revamped email subsystem plus admin features. Adds Invitation model and API endpoints (create, list, revoke, resend, validate) and integrates invitation validation into user registration. Replaces old email account model with richer email models (EmailDomain, EmailAccount, EmailAlias, EmailForwardingRule, DNSProviderConfig) and updates models __init__. Refactors email API to manage domains, accounts, aliases, forwarding, DNS providers, spamassassin, Roundcube/Postfix services and queue/log endpoints. Adds several backend services (invitation, DNS provider, DKIM, Dovecot, Postfix, Roundcube, SpamAssassin, permission service) and a DB migration for permissions/invitations. Enhances admin API with permission management and activity dashboard endpoints, extends audit log actions, and adds a permission_required RBAC decorator. Includes frontend additions for settings/invitations UI and miscellaneous config/service updates and tests.
210 lines
8.2 KiB
Python
210 lines
8.2 KiB
Python
from datetime import datetime, timedelta
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
from app import db
|
|
import json
|
|
|
|
|
|
class User(db.Model):
|
|
__tablename__ = 'users'
|
|
|
|
# Role constants
|
|
ROLE_ADMIN = 'admin'
|
|
ROLE_DEVELOPER = 'developer'
|
|
ROLE_VIEWER = 'viewer'
|
|
VALID_ROLES = [ROLE_ADMIN, ROLE_DEVELOPER, ROLE_VIEWER]
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
email = db.Column(db.String(120), unique=True, nullable=False, index=True)
|
|
username = db.Column(db.String(80), unique=True, nullable=False, index=True)
|
|
password_hash = db.Column(db.String(256), nullable=True)
|
|
auth_provider = db.Column(db.String(50), default='local') # local, google, github, oidc, saml
|
|
role = db.Column(db.String(20), default='developer') # 'admin', 'developer', 'viewer'
|
|
permissions = db.Column(db.Text, nullable=True) # JSON per-feature read/write flags
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
last_login_at = db.Column(db.DateTime, nullable=True)
|
|
created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
|
|
|
|
# Account lockout fields
|
|
failed_login_count = db.Column(db.Integer, default=0)
|
|
locked_until = db.Column(db.DateTime, nullable=True)
|
|
|
|
# Two-Factor Authentication fields
|
|
totp_secret = db.Column(db.String(32), nullable=True) # Base32 encoded secret
|
|
totp_enabled = db.Column(db.Boolean, default=False)
|
|
backup_codes = db.Column(db.Text, nullable=True) # JSON array of hashed backup codes
|
|
totp_confirmed_at = db.Column(db.DateTime, nullable=True) # When 2FA was enabled
|
|
|
|
# Relationships
|
|
applications = db.relationship('Application', backref='owner', lazy='dynamic')
|
|
|
|
# Lockout durations (progressive: 5 min, 15 min, 1 hour)
|
|
LOCKOUT_DURATIONS = [5, 15, 60]
|
|
MAX_FAILED_ATTEMPTS = 5
|
|
|
|
# Per-feature permission system
|
|
PERMISSION_FEATURES = [
|
|
'applications', 'databases', 'docker', 'domains', 'files',
|
|
'monitoring', 'backups', 'security', 'email', 'git', 'cron',
|
|
'terminal', 'users', 'settings', 'servers'
|
|
]
|
|
|
|
ROLE_PERMISSION_TEMPLATES = {
|
|
'admin': {f: {'read': True, 'write': True} for f in PERMISSION_FEATURES},
|
|
'developer': {
|
|
'applications': {'read': True, 'write': True},
|
|
'databases': {'read': True, 'write': True},
|
|
'docker': {'read': True, 'write': True},
|
|
'domains': {'read': True, 'write': True},
|
|
'files': {'read': True, 'write': True},
|
|
'email': {'read': True, 'write': True},
|
|
'git': {'read': True, 'write': True},
|
|
'cron': {'read': True, 'write': True},
|
|
'monitoring': {'read': True, 'write': False},
|
|
'backups': {'read': True, 'write': False},
|
|
'security': {'read': True, 'write': False},
|
|
'terminal': {'read': True, 'write': False},
|
|
'servers': {'read': True, 'write': False},
|
|
'users': {'read': False, 'write': False},
|
|
'settings': {'read': False, 'write': False},
|
|
},
|
|
'viewer': {
|
|
'applications': {'read': True, 'write': False},
|
|
'databases': {'read': True, 'write': False},
|
|
'docker': {'read': True, 'write': False},
|
|
'domains': {'read': True, 'write': False},
|
|
'files': {'read': True, 'write': False},
|
|
'email': {'read': True, 'write': False},
|
|
'git': {'read': True, 'write': False},
|
|
'cron': {'read': True, 'write': False},
|
|
'monitoring': {'read': True, 'write': False},
|
|
'backups': {'read': True, 'write': False},
|
|
'security': {'read': True, 'write': False},
|
|
'terminal': {'read': False, 'write': False},
|
|
'users': {'read': False, 'write': False},
|
|
'settings': {'read': False, 'write': False},
|
|
'servers': {'read': True, 'write': False},
|
|
},
|
|
}
|
|
|
|
@property
|
|
def is_locked(self):
|
|
"""Check if account is currently locked."""
|
|
if self.locked_until is None:
|
|
return False
|
|
return datetime.utcnow() < self.locked_until
|
|
|
|
def record_failed_login(self):
|
|
"""Record a failed login attempt and lock account if threshold reached."""
|
|
self.failed_login_count += 1
|
|
if self.failed_login_count >= self.MAX_FAILED_ATTEMPTS:
|
|
# Calculate lockout duration based on how many times they've been locked
|
|
lockout_index = min(
|
|
(self.failed_login_count - self.MAX_FAILED_ATTEMPTS) // self.MAX_FAILED_ATTEMPTS,
|
|
len(self.LOCKOUT_DURATIONS) - 1
|
|
)
|
|
lockout_minutes = self.LOCKOUT_DURATIONS[lockout_index]
|
|
self.locked_until = datetime.utcnow() + timedelta(minutes=lockout_minutes)
|
|
|
|
def reset_failed_login(self):
|
|
"""Reset failed login count after successful login."""
|
|
self.failed_login_count = 0
|
|
self.locked_until = None
|
|
|
|
def set_password(self, password):
|
|
self.password_hash = generate_password_hash(password)
|
|
|
|
def check_password(self, password):
|
|
if not self.password_hash:
|
|
return False
|
|
return check_password_hash(self.password_hash, password)
|
|
|
|
@property
|
|
def is_admin(self):
|
|
"""Check if user has admin role."""
|
|
return self.role == self.ROLE_ADMIN
|
|
|
|
@property
|
|
def is_developer(self):
|
|
"""Check if user has developer role or higher."""
|
|
return self.role in [self.ROLE_ADMIN, self.ROLE_DEVELOPER]
|
|
|
|
@property
|
|
def is_viewer(self):
|
|
"""Check if user has viewer role or higher (all roles)."""
|
|
return self.role in self.VALID_ROLES
|
|
|
|
def has_role(self, *roles):
|
|
"""Check if user has any of the specified roles."""
|
|
return self.role in roles
|
|
|
|
@property
|
|
def has_password(self):
|
|
return self.password_hash is not None
|
|
|
|
def get_permissions(self):
|
|
"""Return resolved permissions: custom if set, otherwise role template."""
|
|
if self.role == self.ROLE_ADMIN:
|
|
return self.ROLE_PERMISSION_TEMPLATES['admin']
|
|
if self.permissions:
|
|
try:
|
|
return json.loads(self.permissions)
|
|
except (json.JSONDecodeError, TypeError):
|
|
pass
|
|
return self.ROLE_PERMISSION_TEMPLATES.get(self.role, {})
|
|
|
|
def set_permissions(self, perms_dict):
|
|
"""Store custom permissions as JSON."""
|
|
self.permissions = json.dumps(perms_dict) if perms_dict else None
|
|
|
|
def has_permission(self, feature, level='read'):
|
|
"""Check if user has a specific permission. Admin always returns True."""
|
|
if self.role == self.ROLE_ADMIN:
|
|
return True
|
|
perms = self.get_permissions()
|
|
feature_perms = perms.get(feature, {})
|
|
return feature_perms.get(level, False)
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'id': self.id,
|
|
'email': self.email,
|
|
'username': self.username,
|
|
'role': self.role,
|
|
'permissions': self.get_permissions(),
|
|
'is_active': self.is_active,
|
|
'totp_enabled': self.totp_enabled,
|
|
'auth_provider': self.auth_provider or 'local',
|
|
'has_password': self.has_password,
|
|
'created_at': self.created_at.isoformat(),
|
|
'updated_at': self.updated_at.isoformat(),
|
|
'last_login_at': self.last_login_at.isoformat() if self.last_login_at else None,
|
|
'created_by': self.created_by
|
|
}
|
|
|
|
def get_backup_codes(self):
|
|
"""Get the list of backup code hashes."""
|
|
if not self.backup_codes:
|
|
return []
|
|
try:
|
|
return json.loads(self.backup_codes)
|
|
except (json.JSONDecodeError, TypeError):
|
|
return []
|
|
|
|
def set_backup_codes(self, codes_hashes):
|
|
"""Store backup code hashes."""
|
|
self.backup_codes = json.dumps(codes_hashes)
|
|
|
|
def use_backup_code(self, code_hash):
|
|
"""Remove a used backup code."""
|
|
codes = self.get_backup_codes()
|
|
if code_hash in codes:
|
|
codes.remove(code_hash)
|
|
self.set_backup_codes(codes)
|
|
return True
|
|
return False
|
|
|
|
def __repr__(self):
|
|
return f'<User {self.username}>'
|