365 lines
14 KiB
Python
365 lines
14 KiB
Python
"""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)
|