File size: 11,686 Bytes
beacc25 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 | #!/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()
|