File size: 6,574 Bytes
97ce675 | 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 | #!/usr/bin/env python3
"""
Get Users Script for Attestation Store
This script retrieves all users from the API and saves them to a CSV file.
"""
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_OUTPUT_FILE = "downloaded_users.csv"
def parse_args():
"""Parse command line arguments"""
parser = argparse.ArgumentParser(description="Download users from Attestation Store")
parser.add_argument("--url", default=DEFAULT_URL, help=f"Base URL of the Attestation Store API (default: {DEFAULT_URL})")
parser.add_argument("--output", default=DEFAULT_OUTPUT_FILE, help=f"Output CSV file (default: {DEFAULT_OUTPUT_FILE})")
parser.add_argument("--include-inactive", action="store_true", help="Include inactive users in the output")
# Authentication options
auth_group = parser.add_argument_group('Authentication options')
auth_group.add_argument("--token", default=os.getenv("SCRIBE_TOKEN"), help="API token for authentication")
args = parser.parse_args()
# Validate that token is provided
if not args.token:
parser.error("--token must be provided or set the SCRIBE_TOKEN environment variable")
return args
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)
if response.status_code == 200:
try:
result = response.json()
token = result.get('access_token')
if token:
print(f"Successfully obtained JWT access token using API token")
return token
else:
print("Error: Token not found in API response")
return None
except json.JSONDecodeError as e:
print(f"Error decoding JSON response: {str(e)}")
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_users(base_url, token, include_inactive=False):
"""Get users from the API"""
api_url = urljoin(base_url, "/v1/users/bulk")
# Add query parameters
if include_inactive:
api_url += "?include_inactive=true"
# Set up headers with JWT token
auth_header = f"Bearer {token}"
headers = {
"Authorization": auth_header,
"Content-Type": "application/json"
}
print(f"Sending request to {api_url}")
try:
response = requests.get(api_url, headers=headers)
if response.status_code == 200:
result = response.json()
users = result.get('users', [])
print(f"Successfully retrieved {len(users)} users")
return users
else:
print(f"Error: API returned status code {response.status_code}")
print(f"Response: {response.text}")
# 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")
return None
except requests.exceptions.RequestException as e:
print(f"Error connecting to API: {str(e)}")
return None
def save_users_to_csv(users, output_file):
"""Save users to CSV file"""
try:
# Make sure we have users to save
if not users:
print("No users to save")
return False
# Create directory if it doesn't exist
os.makedirs(os.path.dirname(output_file) if os.path.dirname(output_file) else '.', exist_ok=True)
# Define CSV fields based on the first user (should have all required fields)
# We want to maintain the same format as users.csv for compatibility
csv_fields = ['username', 'email', 'password', 'require_password_change']
# Write users to CSV file
with open(output_file, 'w', newline='') as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=csv_fields)
writer.writeheader()
for user in users:
# Only include specified fields
row = {
'username': user.get('username', ''),
'email': user.get('email', ''),
'password': '', # Never include passwords
'require_password_change': 'true' if user.get('require_password_change') else 'false'
}
writer.writerow(row)
print(f"Successfully saved {len(users)} users to {output_file}")
return True
except Exception as e:
print(f"Error saving users to CSV: {str(e)}")
return False
def main():
args = parse_args()
# Get JWT token using API token
print(f"Using provided API token to get a JWT token...")
token = get_token_with_api_token(args.url, args.token)
if not token:
print("\nError: Failed to obtain JWT token")
print("Please check your credentials and try again")
sys.exit(1)
# Get users from API
users = get_users(args.url, token, args.include_inactive)
if not users:
print("No users retrieved from API")
sys.exit(1)
# Save users to CSV
success = save_users_to_csv(users, args.output)
if success:
print(f"Users successfully saved to {args.output}")
else:
print("Failed to save users to CSV")
sys.exit(1)
if __name__ == "__main__":
main()
|