shopdb-flask/shopdb/core/models/user.py

"""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"<Permission {self.name}>"

    @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)
    color = db.Column(db.String(20), comment='CSS color for the role badge')

    # Permissions relationship
    permissions = db.relationship(
        'Permission',
        secondary=rolepermissions,
        backref=db.backref('roles', lazy='dynamic')
    )

    def __repr__(self):
        return f"<Role {self.rolename}>"

    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"<User {self.username}>"

    @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)