ske087
90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
"""
|
|
Internal blueprint — service-to-service endpoints.
|
|
|
|
These routes are NOT exposed through nginx to the public internet.
|
|
They are called from the portal when it needs to pre-provision users.
|
|
Protected with a shared secret in the X-Internal-Token header.
|
|
"""
|
|
import os
|
|
import secrets as _secrets
|
|
from flask import Blueprint, request, jsonify, current_app
|
|
|
|
from app.extensions import db, bcrypt
|
|
|
|
internal_bp = Blueprint('internal', __name__, url_prefix='/internal')
|
|
|
|
|
|
def _check_token():
|
|
"""Return True if the request carries a valid internal sync token."""
|
|
expected = current_app.config.get('INTERNAL_SYNC_SECRET', '')
|
|
provided = request.headers.get('X-Internal-Token', '')
|
|
if not expected or not provided:
|
|
return False
|
|
# Constant-time comparison to prevent timing attacks
|
|
return _secrets.compare_digest(expected, provided)
|
|
|
|
|
|
@internal_bp.route('/sync-user', methods=['POST'])
|
|
def sync_user():
|
|
"""
|
|
Create or update a portal-managed user in DigiServer's local DB.
|
|
|
|
Request JSON:
|
|
{
|
|
"username": "<portal username>",
|
|
"role": "admin" | "user"
|
|
}
|
|
"""
|
|
if not _check_token():
|
|
return jsonify({'error': 'forbidden'}), 403
|
|
|
|
data = request.get_json(silent=True) or {}
|
|
username = (data.get('username') or '').strip()
|
|
role_raw = (data.get('role') or 'user').strip()
|
|
role = 'admin' if role_raw == 'admin' else 'user'
|
|
|
|
if not username:
|
|
return jsonify({'error': 'username required'}), 400
|
|
|
|
from app.models.user import User
|
|
try:
|
|
user = User.query.filter_by(username=username).first()
|
|
if user:
|
|
if user.role != role:
|
|
user.role = role
|
|
db.session.commit()
|
|
status = 'updated'
|
|
else:
|
|
pw_hash = bcrypt.generate_password_hash(
|
|
_secrets.token_hex(32)
|
|
).decode('utf-8')
|
|
user = User(username=username, password=pw_hash, role=role)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
status = 'created'
|
|
|
|
return jsonify({'status': status, 'username': username, 'role': role}), 200
|
|
|
|
except Exception as exc:
|
|
db.session.rollback()
|
|
current_app.logger.error('sync-user error: %s', exc)
|
|
return jsonify({'error': 'internal error'}), 500
|
|
|
|
|
|
@internal_bp.route('/users', methods=['GET'])
|
|
def list_users():
|
|
"""Return a list of portal-managed users (for portal Settings UI)."""
|
|
if not _check_token():
|
|
return jsonify({'error': 'forbidden'}), 403
|
|
|
|
from app.models.user import User
|
|
users = User.query.order_by(User.username).all()
|
|
return jsonify([
|
|
{
|
|
'username': u.username,
|
|
'role': u.role,
|
|
'last_login': u.last_login.isoformat() if u.last_login else None,
|
|
}
|
|
for u in users
|
|
]), 200
|