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

    # Predefined permissions
    PERMISSIONS = [
        # Assets
        ('assets.view', 'View assets', 'assets'),
        ('assets.create', 'Create assets', 'assets'),
        ('assets.edit', 'Edit assets', 'assets'),
        ('assets.delete', 'Delete assets', 'assets'),
        # Machines
        ('machines.view', 'View machines', 'machines'),
        ('machines.create', 'Create machines', 'machines'),
        ('machines.edit', 'Edit machines', 'machines'),
        ('machines.delete', 'Delete machines', 'machines'),
        # Computers
        ('computers.view', 'View computers', 'computers'),
        ('computers.create', 'Create computers', 'computers'),
        ('computers.edit', 'Edit computers', 'computers'),
        ('computers.delete', 'Delete computers', 'computers'),
        # Printers
        ('printers.view', 'View printers', 'printers'),
        ('printers.create', 'Create printers', 'printers'),
        ('printers.edit', 'Edit printers', 'printers'),
        ('printers.delete', 'Delete printers', 'printers'),
        # Network
        ('network.view', 'View network devices', 'network'),
        ('network.create', 'Create network devices', 'network'),
        ('network.edit', 'Edit network devices', 'network'),
        ('network.delete', 'Delete network devices', 'network'),
        # Applications
        ('applications.view', 'View applications', 'applications'),
        ('applications.create', 'Create applications', 'applications'),
        ('applications.edit', 'Edit applications', 'applications'),
        ('applications.delete', 'Delete applications', 'applications'),
        # Knowledge Base
        ('kb.view', 'View knowledge base', 'knowledgebase'),
        ('kb.create', 'Create KB articles', 'knowledgebase'),
        ('kb.edit', 'Edit KB articles', 'knowledgebase'),
        ('kb.delete', 'Delete KB articles', 'knowledgebase'),
        # Notifications
        ('notifications.view', 'View notifications', 'notifications'),
        ('notifications.create', 'Create notifications', 'notifications'),
        ('notifications.edit', 'Edit notifications', 'notifications'),
        ('notifications.delete', 'Delete notifications', 'notifications'),
        # USB devices
        ('usb.view', 'View USB devices', 'usb'),
        ('usb.create', 'Create USB devices', 'usb'),
        ('usb.edit', 'Edit USB devices', 'usb'),
        ('usb.delete', 'Delete USB devices', 'usb'),
        # Warranty
        ('warranty.view', 'View warranties', 'warranty'),
        ('warranty.create', 'Create warranties', 'warranty'),
        ('warranty.edit', 'Edit warranties', 'warranty'),
        ('warranty.delete', 'Delete warranties', 'warranty'),
        # Measuring tools
        ('measuringtools.view', 'View measuring tools', 'measuringtools'),
        ('measuringtools.create', 'Create measuring tools', 'measuringtools'),
        ('measuringtools.edit', 'Edit measuring tools', 'measuringtools'),
        ('measuringtools.delete', 'Delete measuring tools', 'measuringtools'),
        # 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(cls):
        """Seed predefined permissions."""
        created = 0
        for name, description, category in cls.PERMISSIONS:
            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


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"<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 Permission.PERMISSIONS]
        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)

    # 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 Permission.PERMISSIONS]

        perms = set()
        for role in self.roles:
            perms.update(role.getpermissionnames())
        return list(perms)