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'),
        # Deliberately separate from audit.view: reading the log and destroying
        # history are different acts, and many people need only the first.
        ('audit.purge', 'Delete audit log entries past the retention period', '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


def guest_grantable_permissions():
    """Which permissions the guest role may be granted.

    Reading is the safe default and covers almost everything a site wants to
    open up: any `.view` or `.read` permission may be handed to guests.

    A write is different - it lets somebody with no account change this
    database, and the audit trail can then say when and from where but never
    who. So a write has to be BLESSED by whoever owns it, through the plugin's
    get_guest_grantable_permissions(). Secure Send is the case that exists:
    creating a one-view link for somebody is the whole point of the page, and
    the page has always been usable by people with no account here.

    Everything else is refused at the door rather than warned about, because a
    checkbox that quietly makes deleting an asset anonymous is not a warning
    anybody reads twice.
    """
    from flask import current_app, has_app_context

    catalog = full_permission_catalog()
    grantable = {name for name, _, _ in catalog
                 if name.endswith('.view') or name.endswith('.read')}

    if not has_app_context():
        return grantable
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        return grantable

    known = {name for name, _, _ in catalog}
    for name, plugin in pm.get_all_plugins().items():
        if not pm.registry.is_enabled(name):
            continue
        try:
            blessed = plugin.get_guest_grantable_permissions() or []
        except Exception:
            if current_app.config.get('DEBUG') or current_app.config.get('TESTING'):
                raise
            current_app.logger.exception(
                'Plugin %s get_guest_grantable_permissions failed', name)
            continue
        # A plugin may only bless permissions that exist and that it owns the
        # naming of; a typo must not silently widen anything.
        grantable |= {p for p in blessed if p in known}
    return grantable


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}>"

    # The role a caller with no account is treated as holding.
    #
    # Guest access used to be a per-feature boolean setting - one for the
    # Secure Send page, one for each kiosk, one proposed for guides - which
    # meant every new question about what a signed-out visitor may see was
    # answered by writing more code. It is a permissions question, and this app
    # already administers permissions per role, so a guest is now simply a role
    # holding no permissions until somebody grants it some.
    GUEST = 'guest'

    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)

    @classmethod
    def guest(cls):
        """The guest role, or None on a site that has not been seeded yet."""
        return cls.query.filter_by(rolename=cls.GUEST).first()

    @classmethod
    def seedguest(cls):
        """Create the guest role if it is missing. Returns True if created.

        Seeded with NO permissions, which is the same behaviour every site has
        today: an unauthenticated request is refused. What a guest may do is a
        decision each site makes in the role grid, not one it inherits from an
        upgrade.
        """
        if cls.query.filter_by(rolename=cls.GUEST).first() is not None:
            return False
        db.session.add(cls(
            rolename=cls.GUEST,
            description=('Anyone who is not signed in. Holds nothing by '
                         'default; grant it a permission to open that up to '
                         'people with no account.'),
            color='#6C757D',
        ))
        return True

    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)