from flask import Blueprint, request, jsonify, session from app.services.nextcloud import NextcloudClient from app.services import tech_users 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 # ── Nextcloud user endpoints (existing) ────────────────────────────────── @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 # ── Tech user endpoints ────────────────────────────────────────────────── @bp.route('/tech-users', methods=['GET']) def list_tech_users(): """List all tech users (admin only).""" if not _check_admin(): return jsonify({'error': 'Admin privileges required'}), 403 users = tech_users.list_all() # Return as list with username included result = [] for username, data in users.items(): result.append({ 'username': username, 'display_name': data.get('display_name', ''), 'nc_password': data.get('nc_password', ''), 'enabled': data.get('enabled', True), 'created_at': data.get('created_at', ''), 'created_by': data.get('created_by', ''), }) return jsonify({'success': True, 'tech_users': result}), 200 @bp.route('/tech-users', methods=['POST']) def create_tech_user(): """Create a tech user: generate NC password, hash PIN, create NC account.""" if not _check_admin(): return jsonify({'error': 'Admin privileges required'}), 403 data = request.get_json() if not data or 'username' not in data or 'pin' not in data: return jsonify({'error': 'Username and PIN required'}), 400 username = data['username'].strip() display_name = data.get('display_name', '').strip() or username pin = data['pin'] if not username or not pin: return jsonify({'error': 'Username and PIN cannot be empty'}), 400 if len(pin) < 4: return jsonify({'error': 'PIN must be at least 4 digits'}), 400 # Create tech user record (generates NC password) try: record = tech_users.create(username, display_name, pin, session['username']) except ValueError as e: return jsonify({'error': str(e)}), 409 # Create Nextcloud account with the generated password nc_client = _get_nc_client() if not nc_client: # Roll back: remove from JSON tech_users.delete(username) return jsonify({'error': 'Not authenticated'}), 401 nc_result = nc_client.ocs_create_user( username=username, password=record['nc_password'], displayname=display_name, ) if not nc_result.get('success'): # Roll back: remove from JSON tech_users.delete(username) return jsonify({'error': nc_result.get('error', 'Failed to create Nextcloud account')}), 500 return jsonify({ 'success': True, 'username': username, 'display_name': display_name, 'nc_password': record['nc_password'], 'enabled': True, }), 201 @bp.route('/tech-users/', methods=['PUT']) def update_tech_user(username): """Update tech user: display_name, pin, enabled.""" if not _check_admin(): return jsonify({'error': 'Admin privileges required'}), 403 data = request.get_json() if not data: return jsonify({'error': 'No data provided'}), 400 fields = {} if 'display_name' in data: fields['display_name'] = data['display_name'] if 'pin' in data: if len(data['pin']) < 4: return jsonify({'error': 'PIN must be at least 4 digits'}), 400 fields['pin'] = data['pin'] if 'enabled' in data: fields['enabled'] = bool(data['enabled']) try: tech_users.update(username, **fields) except ValueError as e: return jsonify({'error': str(e)}), 404 # Sync enable/disable to Nextcloud if 'enabled' in fields: nc_client = _get_nc_client() if nc_client: if fields['enabled']: nc_client.ocs_enable_user(username) else: nc_client.ocs_disable_user(username) return jsonify({'success': True}), 200 @bp.route('/tech-users//reset-password', methods=['POST']) def reset_tech_user_password(username): """Generate a new NC password, update JSON and Nextcloud.""" if not _check_admin(): return jsonify({'error': 'Admin privileges required'}), 403 try: new_password = tech_users.reset_nc_password(username) except ValueError as e: return jsonify({'error': str(e)}), 404 nc_client = _get_nc_client() if nc_client: result = nc_client.ocs_set_password(username, new_password) if not result.get('success'): return jsonify({'error': result.get('error', 'Failed to update Nextcloud password')}), 500 return jsonify({'success': True, 'nc_password': new_password}), 200 @bp.route('/tech-users/', methods=['DELETE']) def delete_tech_user(username): """Delete tech user from JSON and disable NC account.""" if not _check_admin(): return jsonify({'error': 'Admin privileges required'}), 403 try: tech_users.delete(username) except ValueError as e: return jsonify({'error': str(e)}), 404 # Disable (not delete) the Nextcloud account nc_client = _get_nc_client() if nc_client: nc_client.ocs_disable_user(username) return jsonify({'success': True}), 200