Spaces:

dn-scribe
/

attstore

Runtime error

attstore / scripts /create_batch_users.py

feat: user management: password rest, bulk generation of users, deactivate, delete

beacc25 about 1 year ago

11.7 kB

#!/usr/bin/env python3
"""
Batch User Creation Script for Attestation Store
This script demonstrates creating multiple users via the API.
It reads user data from a CSV file andef get_token_with_api_token(base_url, api_token):
"""

import argparse
import csv
import os
import sys
import requests
import json
from urllib.parse import urljoin

DEFAULT_URL = "http://127.0.0.1:5003"
DEFAULT_CSV_FILE = "scripts/users.csv"

def parse_args():
    """Parse command line arguments"""
    parser = argparse.ArgumentParser(description="Create batch users in Attestation Store")
    parser.add_argument("--url", default=DEFAULT_URL, help=f"Base URL of the Attestation Store API (default: {DEFAULT_URL})")
    parser.add_argument("--file", default=DEFAULT_CSV_FILE, help=f"CSV file containing user data (default: {DEFAULT_CSV_FILE})")
    
    # Authentication options
    auth_group = parser.add_argument_group('Authentication options (one is required)')
    auth_group.add_argument("--username", help="Admin username for authentication")
    auth_group.add_argument("--password", help="Admin password for authentication")
    auth_group.add_argument("--token", default=os.getenv("SCRIBE_TOKEN"), help="API token for authentication (alternative to username/password)")
    
    args = parser.parse_args()
    
    # Validate that either (username and password) or token is provided
    if not ((args.username and args.password) or args.token):
        parser.error("Either --username and --password OR --token must be provided")
    
    return args

def read_users_from_csv(filename):
    """Read user data from CSV file"""
    users = []
    try:
        with open(filename, 'r') as csvfile:
            reader = csv.DictReader(csvfile)
            for row in reader:
                # Validate required fields
                if 'username' not in row or not row['username']:
                    print(f"Warning: Skipping row with missing username: {row}")
                    continue
                
                # Create a user object with defaults for missing fields
                user = {
                    'username': row['username'],
                    'email': row.get('email', f"{row['username']}@example.com"),
                    'password': row.get('password', ''),
                    'role': row.get('role', 'user'),
                    'active': row.get('active', 'true').lower() == 'true',
                    'require_password_change': row.get('require_password_change', 'false').lower() == 'true'
                }
                
                users.append(user)
    except FileNotFoundError:
        print(f"Error: CSV file '{filename}' not found")
        sys.exit(1)
    except Exception as e:
        print(f"Error reading CSV file: {str(e)}")
        sys.exit(1)
    
    return users

def create_users(base_url, users, token):
    """Create users in batch via API"""
    api_url = urljoin(base_url, "/v1/users/bulk")
    
    # Always use Bearer prefix for authorization
    auth_header = f"Bearer {token}"
    
    headers = {
        "Authorization": auth_header,
        "Content-Type": "application/json"
    }
    
    # Format data for the API - API expects a list of users, not a dictionary
    payload = users
    
    print(f"Sending request to {api_url}")
    print(f"Using Authorization: Bearer {token[:5]}...{token[-5:] if len(token) > 10 else ''}")
    
    try:
        response = requests.post(api_url, json=payload, headers=headers)
        
        if response.status_code == 200 or response.status_code == 201:
            result = response.json()
            print(f"Successfully created {len(result.get('success', []))} users")
            
            # Show details of created users
            for user in result.get('success', []):
                print(f"  - Created: {user.get('username')} ({user.get('email')})")
            
            # Show any errors
            for error in result.get('failed', []):
                username = error.get('data', {}).get('username', 'Unknown user')
                error_msg = error.get('error', 'Unknown error')
                print(f"  - Error: {username}: {error_msg}")
                
            return True
        else:
            print(f"Error: API returned status code {response.status_code}")
            print(f"Response: {response.text}")
            print(f"\nFull request details:")
            print(f"URL: {api_url}")
            
            # Additional troubleshooting info for common errors
            if response.status_code == 401:
                print("\nThis looks like an authentication error. Possible causes:")
                print("  - The token may have expired")
                print("  - The token may not be valid")
            elif response.status_code == 403:
                print("\nThis looks like a permission error. Possible causes:")
                print("  - The authenticated user may not have admin privileges")
            elif response.status_code == 422 and "Not enough segments" in response.text:
                print("\nThis is a JWT validation error.")
                print("The bulk user creation endpoint requires a valid JWT token (not an API key).")
                print("Please try again with:")
                print("  - Use the --username and --password options to log in directly")
                print("  - Use a valid JWT token obtained from logging in, not an API key")
            elif response.status_code == 422:
                print("\nThis looks like a validation error. Possible causes:")
                print("  - The request payload format may be incorrect")
                print("  - If using a token from SCRIBE_TOKEN, make sure it's a JWT token, not an API key")
            
            return False
            
    except requests.exceptions.RequestException as e:
        print(f"Error connecting to API: {str(e)}")
        return False

def generate_sample_csv(filename):
    """Generate a sample CSV file if it doesn't exist"""
    if os.path.exists(filename):
        return
    
    # Create directory if it doesn't exist
    os.makedirs(os.path.dirname(filename), exist_ok=True)
    
    with open(filename, 'w', newline='') as csvfile:
        writer = csv.writer(csvfile)
        writer.writerow(['username', 'email', 'password', 'role', 'active', 'require_password_change'])
        writer.writerow(['user1', 'user1@example.com', 'password123', 'user', 'true', 'true'])
        writer.writerow(['user2', 'user2@example.com', 'password456', 'user', 'true', 'false'])
        writer.writerow(['user3', 'user3@example.com', '', 'user', 'true', 'true'])
    
    print(f"Generated sample CSV file: {filename}")

def get_token_with_api_token(base_url, api_token):
    """Get a JWT access token from the API using an API token"""
    login_url = urljoin(base_url, "/v1/login")
    
    # For API token auth, we use a different payload format
    payload = {
        "api_token": api_token
    }
    
    headers = {
        "Content-Type": "application/json"
    }
    
    try:
        print(f"Sending login request to {login_url} using API token...")
        response = requests.post(login_url, json=payload, headers=headers)
        
        # Print raw response for debugging
        print(f"Status code: {response.status_code}")
        print(f"Raw response content: {response.content.decode('utf-8', errors='replace')}")
        
        if response.status_code == 200:
            try:
                # Try to parse JSON response
                result = response.json()
                print(f"JSON parsed successfully: {result}")
                
                token = result.get('access_token')
                if token:
                    print(f"Successfully obtained JWT access token using API token")
                    # Verify it's actually a JWT token (should have 2 dots)
                    if token.count('.') < 2:
                        print("Warning: The obtained token does not appear to be a valid JWT token")
                    return token
                else:
                    print("Error: Token not found in API response")
                    print(f"Response keys: {result.keys()}")
                    
                    # Check for alternative token field names
                    if 'token' in result:
                        token = result.get('token')
                        print(f"Found token in 'token' field instead")
                        return token
                    
                    return None
            except json.JSONDecodeError as e:
                print(f"Error decoding JSON response: {str(e)}")
                print(f"Response might not be valid JSON: {response.text[:100]}...")
                return None
        else:
            print(f"Error: Login API returned status code {response.status_code}")
            print(f"Response: {response.text}")
            return None
            
    except requests.exceptions.RequestException as e:
        print(f"Error connecting to API: {str(e)}")
        return None

def get_token(base_url, username, password):
    """Get a JWT access token from the API using username and password"""
    login_url = urljoin(base_url, "/v1/login")
    
    payload = {
        "username": username,
        "password": password
    }
    
    try:
        print(f"Sending login request to {login_url}...")
        response = requests.post(login_url, json=payload)
        
        if response.status_code == 200:
            result = response.json()
            token = result.get('access_token')
            if token:
                print(f"Successfully obtained JWT access token")
                # Verify it's actually a JWT token (should have 2 dots)
                if token.count('.') < 2:
                    print("Warning: The obtained token does not appear to be a valid JWT token")
                return token
            else:
                print("Error: Token not found in API response")
                print(f"Response: {json.dumps(result, indent=2)}")
                return None
        else:
            print(f"Error: Login API returned status code {response.status_code}")
            print(f"Response: {response.text}")
            return None
            
    except requests.exceptions.RequestException as e:
        print(f"Error connecting to API: {str(e)}")
        return None

def main():
    args = parse_args()
    
    # Get JWT token using either username/password or API token
    token = None
    
    if args.token:
        # Use API token to get JWT token
        print(f"Using provided API token to get a JWT token...")
        token = get_token_with_api_token(args.url, args.token)
    elif args.username and args.password:
        # Use username/password to get JWT token
        print(f"Logging in as {args.username} to get JWT token...")
        token = get_token(args.url, args.username, args.password)
    
    if not token:
        print("\nError: Failed to obtain JWT token")
        print("Please check your credentials and try again")
        sys.exit(1)
    
    # Generate sample CSV if file doesn't exist
    generate_sample_csv(args.file)
    
    # Read users from CSV
    users = read_users_from_csv(args.file)
    if not users:
        print("No valid users found in CSV file")
        sys.exit(1)
    
    print(f"Found {len(users)} users in CSV file")
    
    # Create users via API
    success = create_users(args.url, users, token)
    if success:
        print("Batch user creation completed successfully")
    else:
        print("Batch user creation failed")
        sys.exit(1)

if __name__ == "__main__":
    main()

	#!/usr/bin/env python3
	"""
	Batch User Creation Script for Attestation Store

	This script demonstrates creating multiple users via the API.
	It reads user data from a CSV file andef get_token_with_api_token(base_url, api_token):
	"""

	import argparse
	import csv
	import os
	import sys
	import requests
	import json
	from urllib.parse import urljoin

	DEFAULT_URL = "http://127.0.0.1:5003"
	DEFAULT_CSV_FILE = "scripts/users.csv"

	def parse_args():
	"""Parse command line arguments"""
	parser = argparse.ArgumentParser(description="Create batch users in Attestation Store")
	parser.add_argument("--url", default=DEFAULT_URL, help=f"Base URL of the Attestation Store API (default: {DEFAULT_URL})")
	parser.add_argument("--file", default=DEFAULT_CSV_FILE, help=f"CSV file containing user data (default: {DEFAULT_CSV_FILE})")

	# Authentication options
	auth_group = parser.add_argument_group('Authentication options (one is required)')
	auth_group.add_argument("--username", help="Admin username for authentication")
	auth_group.add_argument("--password", help="Admin password for authentication")
	auth_group.add_argument("--token", default=os.getenv("SCRIBE_TOKEN"), help="API token for authentication (alternative to username/password)")

	args = parser.parse_args()

	# Validate that either (username and password) or token is provided
	if not ((args.username and args.password) or args.token):
	parser.error("Either --username and --password OR --token must be provided")

	return args

	def read_users_from_csv(filename):
	"""Read user data from CSV file"""
	users = []
	try:
	with open(filename, 'r') as csvfile:
	reader = csv.DictReader(csvfile)
	for row in reader:
	# Validate required fields
	if 'username' not in row or not row['username']:
	print(f"Warning: Skipping row with missing username: {row}")
	continue

	# Create a user object with defaults for missing fields
	user = {
	'username': row['username'],
	'email': row.get('email', f"{row['username']}@example.com"),
	'password': row.get('password', ''),
	'role': row.get('role', 'user'),
	'active': row.get('active', 'true').lower() == 'true',
	'require_password_change': row.get('require_password_change', 'false').lower() == 'true'
	}

	users.append(user)
	except FileNotFoundError:
	print(f"Error: CSV file '{filename}' not found")
	sys.exit(1)
	except Exception as e:
	print(f"Error reading CSV file: {str(e)}")
	sys.exit(1)

	return users

	def create_users(base_url, users, token):
	"""Create users in batch via API"""
	api_url = urljoin(base_url, "/v1/users/bulk")

	# Always use Bearer prefix for authorization
	auth_header = f"Bearer {token}"

	headers = {
	"Authorization": auth_header,
	"Content-Type": "application/json"
	}

	# Format data for the API - API expects a list of users, not a dictionary
	payload = users

	print(f"Sending request to {api_url}")
	print(f"Using Authorization: Bearer {token[:5]}...{token[-5:] if len(token) > 10 else ''}")

	try:
	response = requests.post(api_url, json=payload, headers=headers)

	if response.status_code == 200 or response.status_code == 201:
	result = response.json()
	print(f"Successfully created {len(result.get('success', []))} users")

	# Show details of created users
	for user in result.get('success', []):
	print(f" - Created: {user.get('username')} ({user.get('email')})")

	# Show any errors
	for error in result.get('failed', []):
	username = error.get('data', {}).get('username', 'Unknown user')
	error_msg = error.get('error', 'Unknown error')
	print(f" - Error: {username}: {error_msg}")

	return True
	else:
	print(f"Error: API returned status code {response.status_code}")
	print(f"Response: {response.text}")
	print(f"\nFull request details:")
	print(f"URL: {api_url}")

	# Additional troubleshooting info for common errors
	if response.status_code == 401:
	print("\nThis looks like an authentication error. Possible causes:")
	print(" - The token may have expired")
	print(" - The token may not be valid")
	elif response.status_code == 403:
	print("\nThis looks like a permission error. Possible causes:")
	print(" - The authenticated user may not have admin privileges")
	elif response.status_code == 422 and "Not enough segments" in response.text:
	print("\nThis is a JWT validation error.")
	print("The bulk user creation endpoint requires a valid JWT token (not an API key).")
	print("Please try again with:")
	print(" - Use the --username and --password options to log in directly")
	print(" - Use a valid JWT token obtained from logging in, not an API key")
	elif response.status_code == 422:
	print("\nThis looks like a validation error. Possible causes:")
	print(" - The request payload format may be incorrect")
	print(" - If using a token from SCRIBE_TOKEN, make sure it's a JWT token, not an API key")

	return False

	except requests.exceptions.RequestException as e:
	print(f"Error connecting to API: {str(e)}")
	return False

	def generate_sample_csv(filename):
	"""Generate a sample CSV file if it doesn't exist"""
	if os.path.exists(filename):
	return

	# Create directory if it doesn't exist
	os.makedirs(os.path.dirname(filename), exist_ok=True)

	with open(filename, 'w', newline='') as csvfile:
	writer = csv.writer(csvfile)
	writer.writerow(['username', 'email', 'password', 'role', 'active', 'require_password_change'])
	writer.writerow(['user1', 'user1@example.com', 'password123', 'user', 'true', 'true'])
	writer.writerow(['user2', 'user2@example.com', 'password456', 'user', 'true', 'false'])
	writer.writerow(['user3', 'user3@example.com', '', 'user', 'true', 'true'])

	print(f"Generated sample CSV file: {filename}")

	def get_token_with_api_token(base_url, api_token):
	"""Get a JWT access token from the API using an API token"""
	login_url = urljoin(base_url, "/v1/login")

	# For API token auth, we use a different payload format
	payload = {
	"api_token": api_token
	}

	headers = {
	"Content-Type": "application/json"
	}

	try:
	print(f"Sending login request to {login_url} using API token...")
	response = requests.post(login_url, json=payload, headers=headers)

	# Print raw response for debugging
	print(f"Status code: {response.status_code}")
	print(f"Raw response content: {response.content.decode('utf-8', errors='replace')}")

	if response.status_code == 200:
	try:
	# Try to parse JSON response
	result = response.json()
	print(f"JSON parsed successfully: {result}")

	token = result.get('access_token')
	if token:
	print(f"Successfully obtained JWT access token using API token")
	# Verify it's actually a JWT token (should have 2 dots)
	if token.count('.') < 2:
	print("Warning: The obtained token does not appear to be a valid JWT token")
	return token
	else:
	print("Error: Token not found in API response")
	print(f"Response keys: {result.keys()}")

	# Check for alternative token field names
	if 'token' in result:
	token = result.get('token')
	print(f"Found token in 'token' field instead")
	return token

	return None
	except json.JSONDecodeError as e:
	print(f"Error decoding JSON response: {str(e)}")
	print(f"Response might not be valid JSON: {response.text[:100]}...")
	return None
	else:
	print(f"Error: Login API returned status code {response.status_code}")
	print(f"Response: {response.text}")
	return None

	except requests.exceptions.RequestException as e:
	print(f"Error connecting to API: {str(e)}")
	return None

	def get_token(base_url, username, password):
	"""Get a JWT access token from the API using username and password"""
	login_url = urljoin(base_url, "/v1/login")

	payload = {
	"username": username,
	"password": password
	}

	try:
	print(f"Sending login request to {login_url}...")
	response = requests.post(login_url, json=payload)

	if response.status_code == 200:
	result = response.json()
	token = result.get('access_token')
	if token:
	print(f"Successfully obtained JWT access token")
	# Verify it's actually a JWT token (should have 2 dots)
	if token.count('.') < 2:
	print("Warning: The obtained token does not appear to be a valid JWT token")
	return token
	else:
	print("Error: Token not found in API response")
	print(f"Response: {json.dumps(result, indent=2)}")
	return None
	else:
	print(f"Error: Login API returned status code {response.status_code}")
	print(f"Response: {response.text}")
	return None

	except requests.exceptions.RequestException as e:
	print(f"Error connecting to API: {str(e)}")
	return None

	def main():
	args = parse_args()

	# Get JWT token using either username/password or API token
	token = None

	if args.token:
	# Use API token to get JWT token
	print(f"Using provided API token to get a JWT token...")
	token = get_token_with_api_token(args.url, args.token)
	elif args.username and args.password:
	# Use username/password to get JWT token
	print(f"Logging in as {args.username} to get JWT token...")
	token = get_token(args.url, args.username, args.password)

	if not token:
	print("\nError: Failed to obtain JWT token")
	print("Please check your credentials and try again")
	sys.exit(1)

	# Generate sample CSV if file doesn't exist
	generate_sample_csv(args.file)

	# Read users from CSV
	users = read_users_from_csv(args.file)
	if not users:
	print("No valid users found in CSV file")
	sys.exit(1)

	print(f"Found {len(users)} users in CSV file")

	# Create users via API
	success = create_users(args.url, users, token)
	if success:
	print("Batch user creation completed successfully")
	else:
	print("Batch user creation failed")
	sys.exit(1)

	if __name__ == "__main__":
	main()