mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-06 12:02:51 +09:00
Juan Denis
c8c93a8e0c
Add session-family revocation and migration 097, require current credentials for password changes, constrain API keys, and require verified WebAuthn authentication. Update browser token handling and add boundary regressions. Existing JWTs require fresh login; restricted keys deny undeclared endpoints.
255 lines
10 KiB
Python
255 lines
10 KiB
Python
from datetime import datetime, timedelta
|
|
from sqlalchemy.orm import validates
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
from app import db
|
|
from app.models.json_column_mixin import JsonColumnMixin
|
|
import json
|
|
import secrets
|
|
|
|
|
|
class User(JsonColumnMixin, 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', index=True) # 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)
|
|
auth_version = db.Column(db.String(32), nullable=False,
|
|
default=lambda: secrets.token_hex(16), server_default='0')
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True)
|
|
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, index=True)
|
|
|
|
@validates('email')
|
|
def _normalize_email(self, key, email):
|
|
"""Store emails in lowercase for consistent matching."""
|
|
return email.lower() if isinstance(email, str) else email
|
|
|
|
# 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
|
|
|
|
# Sidebar preferences: { preset: 'full'|'web'|'email'|'devops'|'minimal'|'custom', hiddenItems: [...] }
|
|
sidebar_config = db.Column(db.Text, nullable=True)
|
|
|
|
# Preferred UI language as a BCP-47 language code (plan 79 B2). NULL means
|
|
# "follow the panel default", which is a genuinely different state from an
|
|
# explicit choice of English -- the same distinction ThemeContext draws
|
|
# between an inherited skin and a picked one.
|
|
language = db.Column(db.String(10), nullable=True)
|
|
|
|
# Per-user Setup Health item snoozes: JSON { item_key: iso_expiry } (plan 22
|
|
# Phase 6). Panel-item snoozes live in a SettingsService map instead.
|
|
setup_snoozes = db.Column(db.Text, nullable=True)
|
|
|
|
# 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)
|
|
self.revoke_sessions()
|
|
|
|
@validates('is_active')
|
|
def _invalidate_disabled_sessions(self, key, active):
|
|
if not active:
|
|
self.revoke_sessions()
|
|
return active
|
|
|
|
def revoke_sessions(self):
|
|
"""Invalidate access/refresh/MFA tokens and unused login links."""
|
|
self.auth_version = secrets.token_hex(16)
|
|
if self.id is not None:
|
|
from app.models.login_link import LoginLink
|
|
LoginLink.query.filter(
|
|
(LoginLink.user_id == self.id) | (LoginLink.created_by_id == self.id)
|
|
).delete(synchronize_session=False)
|
|
|
|
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']
|
|
custom = self._json_read('permissions', None)
|
|
if custom is not None:
|
|
return custom
|
|
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 get_sidebar_config(self):
|
|
"""Return sidebar config dict, or default."""
|
|
return self._json_read('sidebar_config',
|
|
{'preset': 'full', 'hiddenItems': []})
|
|
|
|
def set_sidebar_config(self, config):
|
|
"""Store sidebar config as JSON."""
|
|
self._json_write('sidebar_config', config)
|
|
|
|
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,
|
|
'passkey_enabled': self.passkeys.filter_by(is_active=True).count() > 0,
|
|
'sidebar_config': self.get_sidebar_config(),
|
|
'language': self.language,
|
|
'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,
|
|
'is_admin': self.is_admin
|
|
}
|
|
|
|
def get_backup_codes(self):
|
|
"""Get the list of backup code hashes."""
|
|
return self._json_read('backup_codes', [])
|
|
|
|
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}>'
|