shopdb-flask/shopdb/core/models/apitoken.py

"""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)
    # JSON array of RESOURCE-scope names (a different axis from `scopes`, which
    # is what the token may DO). Today these are geenforce manifest scope names
    # (e.g. gea-shopfloor-display): a resource-bound geenforce.fetch token may
    # only pull those scopes' manifests and only blobs those manifests
    # reference. NULL = unrestricted (any resource), for back-compat. See
    # resourcescopelist below.
    resourcescopes = 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))

    @property
    def resourcescopelist(self):
        """Parsed resource-scope names, or None when the token is unrestricted."""
        if self.resourcescopes is None:
            return None
        try:
            value = json.loads(self.resourcescopes)
        except (ValueError, TypeError):
            return None
        return value if isinstance(value, list) else None

    @resourcescopelist.setter
    def resourcescopelist(self, names):
        """Store a resource-scope list, or None to clear the restriction."""
        if names is None:
            self.resourcescopes = None
        else:
            self.resourcescopes = 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,
            'resourcescopes': self.resourcescopelist,
            '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