"""User and authentication models.""" from datetime import datetime, timezone from shopdb.extensions import db from .base import BaseModel # Association table for user roles (many-to-many) userroles = db.Table( 'userroles', db.Column('userid', db.Integer, db.ForeignKey('users.userid'), primary_key=True), db.Column('roleid', db.Integer, db.ForeignKey('roles.roleid'), primary_key=True) ) # Association table for role permissions (many-to-many) rolepermissions = db.Table( 'rolepermissions', db.Column('roleid', db.Integer, db.ForeignKey('roles.roleid'), primary_key=True), db.Column('permissionid', db.Integer, db.ForeignKey('permissions.permissionid'), primary_key=True) ) class Permission(db.Model): """ Permission model for granular access control. Permissions are predefined and assigned to roles. """ __tablename__ = 'permissions' permissionid = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(50), unique=True, nullable=False) description = db.Column(db.String(255)) category = db.Column(db.String(50), default='general') # For grouping in UI # Core permission catalog. This is CORE ONLY - permissions for a plugin # domain (machines, computers, printers, network, knowledgebase, # notifications, usb, warranty, measuringtools) live in that plugin's # BasePlugin.get_permissions() hook, not here (plugin-is-the-product). The # merged core+plugins view is full_permission_catalog() below. CORE_PERMISSIONS = [ # Assets ('assets.view', 'View assets', 'assets'), ('assets.create', 'Create assets', 'assets'), ('assets.edit', 'Edit assets', 'assets'), ('assets.delete', 'Delete assets', 'assets'), # Applications ('applications.view', 'View applications', 'applications'), ('applications.create', 'Create applications', 'applications'), ('applications.edit', 'Edit applications', 'applications'), ('applications.delete', 'Delete applications', 'applications'), # Reports ('reports.view', 'View reports', 'reports'), ('reports.export', 'Export reports', 'reports'), # Settings ('settings.view', 'View settings', 'admin'), ('settings.edit', 'Edit settings', 'admin'), # Users ('users.view', 'View users', 'admin'), ('users.create', 'Create users', 'admin'), ('users.edit', 'Edit users', 'admin'), ('users.delete', 'Delete users', 'admin'), # Audit ('audit.view', 'View audit logs', 'admin'), # API tokens ('apitokens.create', 'Create and manage API tokens', 'apitokens'), # Collector service tokens. A token scoped to ONLY this permission is a # collector service token: it authorizes the collector ingest API and # nothing else (scoped tokens pass require_permission only for listed # perms, and this perm gates no other route). See collector.py. ('collector.ingest', 'Submit collector payloads (fleet reporting)', 'collector'), ] def __repr__(self): return f"" @classmethod def seed_entries(cls, entries): """Idempotently create Permission rows from (name, desc, category) tuples (or equivalent dicts). Returns how many were created.""" created = 0 for name, description, category in normalize_permission_entries(entries): if not cls.query.filter_by(name=name).first(): perm = cls(name=name, description=description, category=category) db.session.add(perm) created += 1 return created @classmethod def seed(cls): """Seed the full permission catalog (core plus enabled plugins).""" return cls.seed_entries(full_permission_catalog()) def normalize_permission_entries(entries): """Coerce catalog entries to (name, description, category) tuples. Accepts tuples/lists in that order, or dicts with those keys, so a plugin may return either shape from get_permissions(). """ normalized = [] for entry in entries or []: if isinstance(entry, dict): normalized.append(( entry['name'], entry.get('description', ''), entry.get('category', 'general'), )) else: name, description, category = entry normalized.append((name, description, category)) return normalized def full_permission_catalog(include_disabled=False): """The full permission catalog: core permissions plus every LOADED plugin's get_permissions(). By default only ENABLED plugins contribute, so a disabled plugin's permissions are not offered for new scope grants or role edits (existing Permission rows and role links persist - see the get_permissions docstring). Pass include_disabled=True to include all installed plugins. Follows the dashboard.py consumer pattern: a broken plugin is re-raised in dev/test and isolated (logged) in prod. Returns (name, description, category) tuples. """ from flask import current_app, has_app_context catalog = list(Permission.CORE_PERMISSIONS) if not has_app_context(): return catalog pm = current_app.extensions.get('plugin_manager') if not pm: return catalog for name, plugin in pm.get_all_plugins().items(): if not include_disabled and not pm.registry.is_enabled(name): continue try: catalog.extend( normalize_permission_entries(plugin.get_permissions())) except Exception: if current_app.config.get('DEBUG') or current_app.config.get('TESTING'): raise current_app.logger.exception( 'Plugin %s get_permissions failed', name) return catalog class Role(BaseModel): """User role model.""" __tablename__ = 'roles' roleid = db.Column(db.Integer, primary_key=True) rolename = db.Column(db.String(50), unique=True, nullable=False) description = db.Column(db.Text) # Permissions relationship permissions = db.relationship( 'Permission', secondary=rolepermissions, backref=db.backref('roles', lazy='dynamic') ) def __repr__(self): return f"" def haspermission(self, permission_name: str) -> bool: """Check if role has a specific permission.""" # Admin role has all permissions if self.rolename == 'admin': return True return any(p.name == permission_name for p in self.permissions) def getpermissionnames(self) -> list: """Get list of permission names.""" if self.rolename == 'admin': return [p[0] for p in full_permission_catalog()] return [p.name for p in self.permissions] class User(BaseModel): """User model for authentication.""" __tablename__ = 'users' userid = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(100), unique=True, nullable=False, index=True) email = db.Column(db.String(255), unique=True, nullable=False) passwordhash = db.Column(db.String(255), nullable=False) # Profile firstname = db.Column(db.String(100)) lastname = db.Column(db.String(100)) # Status lastlogindate = db.Column(db.DateTime) failedlogins = db.Column(db.Integer, default=0) lockeduntil = db.Column(db.DateTime) # Forced password change: set true when an admin creates the account, so the # user is steered through a password change before landing in the app. mustchangepassword = db.Column(db.Boolean, default=False, nullable=False) # Relationships roles = db.relationship( 'Role', secondary=userroles, backref=db.backref('users', lazy='dynamic') ) def __repr__(self): return f"" @property def islocked(self): """Check if account is locked.""" if self.lockeduntil: return datetime.now(timezone.utc).replace(tzinfo=None) < self.lockeduntil return False def hasrole(self, rolename: str) -> bool: """Check if user has a specific role.""" return any(r.rolename == rolename for r in self.roles) def haspermission(self, permission_name: str) -> bool: """Check if user has a specific permission through any role.""" # Admin role has all permissions if self.hasrole('admin'): return True return any(r.haspermission(permission_name) for r in self.roles) def getpermissions(self) -> list: """Get list of all permission names from all roles.""" if self.hasrole('admin'): return [p[0] for p in full_permission_catalog()] perms = set() for role in self.roles: perms.update(role.getpermissionnames()) return list(perms)