Plugins declare their own RBAC permissions instead of core accumulating them: 36 permissions moved out of the core catalog into the 9 owning plugins (core keeps the 19 its own blueprints enforce). The catalog is resolved dynamically (core + enabled plugins) and feeds the roles grid, the token scope picker and ceiling, and flask seed permissions; installing or enabling a plugin seeds its permissions automatically. A disabled plugin drops out of the assignable catalog while existing role links keep working. New plugins - bundled or external - now bring their permissions with zero core edits. 781 tests pass; live-verified with a machines.edit-scoped token. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
129 lines
5.4 KiB
Python
129 lines
5.4 KiB
Python
"""Personal API token model.
|
|
|
|
A personal API token (PAT) lets a script or integration authenticate as a
|
|
user without the hourly-expiring login JWT. The secret is shown ONCE at
|
|
creation; only its sha256 hash is stored. By default the token acts as its
|
|
owning user, so the existing role/permission decorators authorize it unchanged.
|
|
A token MAY optionally carry a scopes list (see scopes/scopelist): a scoped
|
|
token grants only the listed permissions, intersected with what the owner holds,
|
|
and suspends the admin bypass. See shopdb/utils/authz.py for the enforcement.
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
|
|
from shopdb.extensions import db
|
|
from .base import BaseModel
|
|
|
|
|
|
def _utcnow():
|
|
# naive UTC to match the other DB DateTime columns (stored without tzinfo)
|
|
return datetime.now(timezone.utc).replace(tzinfo=None)
|
|
|
|
|
|
# Wire label on the full secret. Scripts send "Authorization: Bearer <secret>".
|
|
TOKEN_SECRET_PREFIX = 'shopdb_pat_'
|
|
# Hex chars of randomness after the label (secrets.token_hex(20) => 40 hex).
|
|
_TOKEN_RANDOM_BYTES = 20
|
|
# How many leading random-hex chars we keep in the clear for display/lookup.
|
|
_TOKEN_PREFIX_LEN = 8
|
|
|
|
|
|
class ApiToken(BaseModel):
|
|
"""Personal API token. Stores only the hash of the secret."""
|
|
__tablename__ = 'apitokens'
|
|
|
|
tokenid = db.Column(db.Integer, primary_key=True)
|
|
# The token acts as this user; NOT NULL so authz always has a principal.
|
|
userid = db.Column(db.Integer, db.ForeignKey('users.userid'),
|
|
nullable=False, index=True)
|
|
# What the token is for (e.g. "legacy import runner").
|
|
name = db.Column(db.String(100), nullable=False)
|
|
# First few random-hex chars, kept clear so a user can tell tokens apart.
|
|
tokenprefix = db.Column(db.String(16), nullable=True, index=True)
|
|
# sha256 hex of the full secret. Unique so a hash lookup finds one row.
|
|
tokenhash = db.Column(db.String(64), unique=True, nullable=False, index=True)
|
|
# Null expiresat means the token never expires.
|
|
expiresat = db.Column(db.DateTime, nullable=True)
|
|
# Last time the token authenticated a request (throttled write).
|
|
lastusedat = db.Column(db.DateTime, nullable=True)
|
|
# JSON array of permission-name strings. NULL = unscoped (full owner
|
|
# authority). A scoped token grants ONLY these, intersected with what the
|
|
# owner holds, and suspends the admin bypass. See scopelist below.
|
|
scopes = db.Column(db.Text, nullable=True)
|
|
|
|
user = db.relationship('User', backref=db.backref('apitokens', lazy='dynamic'))
|
|
|
|
@staticmethod
|
|
def generate_secret() -> str:
|
|
"""Return a fresh full secret: shopdb_pat_<40 hex>. Never stored."""
|
|
return TOKEN_SECRET_PREFIX + secrets.token_hex(_TOKEN_RANDOM_BYTES)
|
|
|
|
@staticmethod
|
|
def hash_secret(secret: str) -> str:
|
|
"""sha256 hex of the full secret. The token has 160 bits of entropy,
|
|
so a plain hash lookup (not a slow password hash) is appropriate."""
|
|
return hashlib.sha256(secret.encode('utf-8')).hexdigest()
|
|
|
|
@staticmethod
|
|
def prefix_of(secret: str) -> str:
|
|
"""The clear display prefix (leading random-hex chars) of a secret."""
|
|
randompart = secret[len(TOKEN_SECRET_PREFIX):]
|
|
return randompart[:_TOKEN_PREFIX_LEN]
|
|
|
|
@property
|
|
def is_expired(self) -> bool:
|
|
"""True when expiresat is set and in the past."""
|
|
return self.expiresat is not None and self.expiresat < _utcnow()
|
|
|
|
@property
|
|
def scopelist(self):
|
|
"""Parsed scope names as a list, or None when the token is unscoped."""
|
|
if self.scopes is None:
|
|
return None
|
|
try:
|
|
value = json.loads(self.scopes)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
return value if isinstance(value, list) else None
|
|
|
|
@scopelist.setter
|
|
def scopelist(self, names):
|
|
"""Store a scope list, or None to clear scoping. Does NOT validate the
|
|
names; callers reject bad input first via unknown_scope_names."""
|
|
if names is None:
|
|
self.scopes = None
|
|
else:
|
|
self.scopes = json.dumps(list(names))
|
|
|
|
@staticmethod
|
|
def unknown_scope_names(names) -> list:
|
|
"""Return the subset of names that are not in the permission catalog.
|
|
|
|
The catalog is core plus every ENABLED plugin's permissions, so a scope
|
|
naming a disabled plugin's permission is treated as unknown."""
|
|
from shopdb.core.models.user import full_permission_catalog
|
|
known = {name for name, _desc, _cat in full_permission_catalog()}
|
|
return [n for n in names if n not in known]
|
|
|
|
def to_dict(self, include_owner: bool = False) -> dict:
|
|
"""Serialize for the API. NEVER includes the hash or the secret."""
|
|
result = {
|
|
'tokenid': self.tokenid,
|
|
'userid': self.userid,
|
|
'name': self.name,
|
|
'tokenprefix': self.tokenprefix,
|
|
'displayprefix': f'{TOKEN_SECRET_PREFIX}{self.tokenprefix or ""}',
|
|
'expiresat': self.expiresat.isoformat() + 'Z' if self.expiresat else None,
|
|
'lastusedat': self.lastusedat.isoformat() + 'Z' if self.lastusedat else None,
|
|
'scopes': self.scopelist,
|
|
'isactive': self.isactive,
|
|
'isexpired': self.is_expired,
|
|
'createddate': self.createddate.isoformat() + 'Z' if self.createddate else None,
|
|
}
|
|
if include_owner:
|
|
result['username'] = self.user.username if self.user else None
|
|
return result
|