Spaces:

dn-scribe
/

attstore

Runtime error

App Files Files Community

attstore / scripts /get_users.py

dn-scribe

feat: download user list

97ce675 about 1 year ago

Raw History Blame Contribute Delete

6.57 kB

#!/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()

	#!/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()