from flask import Blueprint, request, jsonify, session from app.services.nextcloud import NextcloudClient from config import Config import base64 bp = Blueprint('admin', __name__, url_prefix='/api/admin') def _get_nc_client(): """Get authenticated Nextcloud client from session.""" if 'username' not in session or 'password' not in session: return None username = session['username'] password = base64.b64decode(session['password'].encode()).decode() return NextcloudClient(Config.NEXTCLOUD_URL, username, password) def _check_admin(): """Check if current user has admin privileges.""" if not session.get('is_admin', False): return False return True @bp.route('/users', methods=['GET']) def list_users(): """List all Nextcloud users (admin only).""" if not _check_admin(): return jsonify({'error': 'Admin privileges required'}), 403 nc_client = _get_nc_client() if not nc_client: return jsonify({'error': 'Not authenticated'}), 401 result = nc_client.ocs_get_users() if not result.get('success'): return jsonify({'error': result.get('error', 'Failed to list users')}), 500 return jsonify(result), 200 @bp.route('/users', methods=['POST']) def create_user(): """Create a new Nextcloud user (admin only).""" if not _check_admin(): return jsonify({'error': 'Admin privileges required'}), 403 data = request.get_json() if not data or 'username' not in data or 'password' not in data: return jsonify({'error': 'Username and password required'}), 400 username = data['username'].strip() password = data['password'] email = data.get('email', '').strip() displayname = data.get('displayName', '').strip() groups = data.get('groups', []) if not username or not password: return jsonify({'error': 'Username and password cannot be empty'}), 400 nc_client = _get_nc_client() if not nc_client: return jsonify({'error': 'Not authenticated'}), 401 result = nc_client.ocs_create_user( username=username, password=password, email=email if email else None, displayname=displayname if displayname else None, groups=groups if groups else None ) if not result.get('success'): return jsonify({'error': result.get('error', 'Failed to create user')}), 500 return jsonify(result), 201 @bp.route('/users//enable', methods=['PUT']) def enable_user(username): """Enable a user account (admin only).""" if not _check_admin(): return jsonify({'error': 'Admin privileges required'}), 403 nc_client = _get_nc_client() if not nc_client: return jsonify({'error': 'Not authenticated'}), 401 result = nc_client.ocs_enable_user(username) if not result.get('success'): return jsonify({'error': result.get('error', 'Failed to enable user')}), 500 return jsonify(result), 200 @bp.route('/users//disable', methods=['PUT']) def disable_user(username): """Disable a user account (admin only).""" if not _check_admin(): return jsonify({'error': 'Admin privileges required'}), 403 nc_client = _get_nc_client() if not nc_client: return jsonify({'error': 'Not authenticated'}), 401 result = nc_client.ocs_disable_user(username) if not result.get('success'): return jsonify({'error': result.get('error', 'Failed to disable user')}), 500 return jsonify(result), 200 @bp.route('/users/', methods=['DELETE']) def delete_user(username): """Delete a user account (admin only).""" if not _check_admin(): return jsonify({'error': 'Admin privileges required'}), 403 # Prevent self-deletion if username == session.get('username'): return jsonify({'error': 'Cannot delete your own account'}), 400 nc_client = _get_nc_client() if not nc_client: return jsonify({'error': 'Not authenticated'}), 401 result = nc_client.ocs_delete_user(username) if not result.get('success'): return jsonify({'error': result.get('error', 'Failed to delete user')}), 500 return jsonify(result), 200