Files
Autoparts-DB/pos/middleware.py
consultoria-as f42910f4f6
Some checks failed
CI / lint-and-test (3.11) (push) Has been cancelled
CI / lint-and-test (3.13) (push) Has been cancelled
feat: cashier/counter reports, service-order & remission flows, Rached migration utils
- Add "Mis cortes de caja" report for cashiers/counters with sales detail.
- Cash register history scoped to own cuts for non-admin roles; new /register/<id>/sales endpoint.
- Remove dashboard from cashier menu; add Reports to cashier/counter.
- Service orders: assign mechanic, budget field, invoice flag, counter/cashier can add items/remissions, convert to remission.
- Remission notes module (UI, CSS, courier, counter remissions).
- Customer hard-delete and vehicle/customer linkage in workshop.
- POS: always show search results, compact payment grid, credit validation, tier pricing (5%/10%), ticket with customer/folio.
- Inventory: CSV template with sku_secondary, alias import.
- Rached migration scripts and DB migrations.
- Version-bump cached JS/CSS query strings.

Excludes local Rached session tokens/captures (rached_*.json / rached_*.txt).
2026-07-02 12:51:56 +00:00

58 lines
2.3 KiB
Python

# /home/Autopartes/pos/middleware.py
"""Auth middleware for POS: JWT validation + tenant resolution + permission checks."""
import jwt
from functools import wraps
from flask import request, jsonify, g
from config import JWT_SECRET
def require_auth(*required_permissions):
"""Decorator: validate JWT, resolve tenant, optionally check permissions.
Usage:
@require_auth() # any authenticated employee
@require_auth('pos.sell') # needs specific permission
@require_auth('pos.sell', 'pos.discount') # needs ALL listed permissions
"""
def decorator(f):
@wraps(f)
def decorated(*args, **kwargs):
auth_header = request.headers.get('Authorization', '')
if not auth_header.startswith('Bearer '):
return jsonify({'error': 'Token required'}), 401
try:
payload = jwt.decode(auth_header[7:], JWT_SECRET, algorithms=['HS256'])
except jwt.ExpiredSignatureError:
return jsonify({'error': 'Token expired'}), 401
except jwt.InvalidTokenError:
return jsonify({'error': 'Invalid token'}), 401
if payload.get('type') not in ('pos_access', 'access'):
return jsonify({'error': 'Invalid token type'}), 401
g.tenant_id = payload['tenant_id']
g.employee_id = payload['employee_id']
g.employee_role = payload['role']
g.employee_name = payload['name']
g.branch_id = payload.get('branch_id')
g.permissions = set(payload.get('permissions', []))
g.device_id = request.headers.get('X-Device-Id', 'unknown')
# Check permissions
if required_permissions:
missing = set(required_permissions) - g.permissions
# owner/admin roles bypass all permission checks
if g.employee_role not in ('owner', 'admin') and missing:
return jsonify({'error': f'Missing permissions: {", ".join(missing)}'}), 403
return f(*args, **kwargs)
return decorated
return decorator
def has_permission(permission):
"""Check if current user has a specific permission. Use inside a route."""
return g.employee_role in ('owner', 'admin') or permission in g.permissions