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'),
        # Deliberately NOT '.view': reading the catalogue and pulling the bytes
        # are different acts. A script body is SYSTEM-context PowerShell and an
        # installer is a vendor binary, so the routes that serve them are gated
        # here rather than on the metadata permission a site may hand to guests.
        ('applications.download',
         'Download application installers and script bodies', '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


# Core permissions a site MAY grant the guest role. Declared, not derived.
# Each one has been read against what its routes actually return.
#
# Deliberately absent, with the reason, so nobody re-adds them by reflex:
#   settings.view  - every setting row, and secret masking is a substring test
#   audit.view     - usernames, addresses, and the before/after of every change
#   users.view     - gates nothing; list_users is require_role('admin')
CORE_GUEST_GRANTABLE = {
    # Aggregate counts only; no row-level data reaches the caller.
    'reports.view',
    # Catalogue metadata. Safe only because the bytes moved to
    # applications.download, which is not grantable.
    'applications.view',
    # Assets, and the global search over them. Safe only because search now
    # gates each domain on its own permission: holding this reaches assets and
    # nothing else, where it used to reach the staff directory and the USB,
    # notification and network domains too.
    'assets.view',
}


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

    Every entry is DECLARED, by core below or by a plugin through
    get_guest_grantable_permissions(). Nothing qualifies by being spelled a
    particular way.

    It used to: any name ending `.view` or `.read` was grantable, on the
    reasoning that reading is the safe default. That is true of a permission
    that gates a list of asset numbers and false of one that gates bytes, and
    the suffix cannot tell them apart. `applications.view` reached the
    SYSTEM-context PowerShell bodies and the installer binaries;
    `backups.view` returned the full payload the separate backups.download
    permission exists to protect; `assets.view` gated a global search that
    queries the staff directory. None of those were decided. They were named.

    So the rule is inverted: a permission is grantable when somebody wrote it
    down, having looked at what it actually serves. Adding one is a code
    change, which is the point - it puts the decision in a diff rather than in
    a suffix.

    Naming one here does not open it. It makes the checkbox available in the
    role grid; a site still has to tick it.

    The bar, for a read: could an anonymous visitor hold this, forever, and
    would the site be content with everything it returns? For a write, add
    that the act can never be attributed to anybody.
    """
    from flask import current_app, has_app_context

    catalog = full_permission_catalog()
    known = {name for name, _, _ in catalog}
    grantable = {name for name in CORE_GUEST_GRANTABLE if name in known}

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

    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)