Hugging Face's logo

Spaces:
dn-scribe
/
attstore
Runtime error

App Files Community
attstore / api /login.py
dn-scribe's picture
dn-scribe
feat: download user list
97ce675
Raw History Blame Contribute Delete
27 kB
from flask import Blueprint, request, jsonify
from flask_jwt_extended import create_access_token, get_jwt_identity, jwt_required
from werkzeug.security import check_password_hash
from datetime import datetime, timedelta, timezone
from models import User, ApiKey
from app import db
from logging_utils import log_event
import base64
login_bp = Blueprint('login', __name__)
def is_admin_user(user_id):
"""Check if user is admin (for now, check if username is 'admin')"""
user = User.query.get(user_id)
return user and user.username == 'admin' and user.is_active
@login_bp.route('/login', methods=['POST'])
def login():
"""User login endpoint - supports multiple authentication methods"""
try:
data = request.get_json()
if not data:
return jsonify({'error': 'JSON data required'}), 400
# Extract different authentication methods
api_token = data.get('api_token')
client_id = data.get('client_id')
client_secret = data.get('client_secret')
username = data.get('username')
password = data.get('password')
# Legacy support
api_key = data.get('api_key')
user = None
auth_method = None
if api_token or api_key:
# Method 1: API Token authentication
token = api_token or api_key
api_key_obj = ApiKey.query.filter_by(key=token).first()
if not api_key_obj or not api_key_obj.is_valid():
log_event(
severity=4,
message=f"Failed login attempt with invalid API token",
event_type="login_failed_api_token",
source_ip=request.remote_addr
)
return jsonify({'error': 'Invalid or expired API token'}), 401
user = User.query.get(api_key_obj.user_id)
auth_method = "api_token"
elif client_id and client_secret:
# Method 2: Client ID/Secret authentication (base64 decoded from API key)
import base64
try:
# Reconstruct base64 credentials
# Try to decode client_secret as base64 first
try:
decoded_secret = base64.b64decode(client_secret).decode('utf-8')
parts = decoded_secret.split(':')
if parts[0] == client_id:
token = client_secret
else:
# Fall back to original method
credentials = f"{client_id}:{client_secret}"
token = base64.b64encode(credentials.encode()).decode()
except Exception:
# If decoding fails, use original method
credentials = f"{client_id}:{client_secret}"
token = base64.b64encode(credentials.encode()).decode()
api_key_obj = ApiKey.query.filter_by(key=token).first()
if not api_key_obj or not api_key_obj.is_valid():
log_event(
severity=4,
message=f"Failed login attempt with invalid client credentials for: {client_id}",
event_type="login_failed_client_creds",
source_ip=request.remote_addr
)
return jsonify({'error': 'Invalid client credentials'}), 401
user = User.query.get(api_key_obj.user_id)
auth_method = "client_credentials"
except Exception as e:
log_event(
severity=4,
message=f"Failed to process client credentials for: {client_id}",
event_type="login_failed_client_creds_format",
source_ip=request.remote_addr
)
return jsonify({'error': 'Invalid client credentials format'}), 401
elif username and password:
# Method 3: Username/password authentication
user = User.query.filter_by(username=username).first()
if not user or not user.check_password(password):
# Record the failed login if user exists
if user:
account_deactivated = user.record_login_failure()
db.session.commit()
if account_deactivated:
log_event(
severity=3,
message=f"Account deactivated after 50 failed login attempts: {username}",
event_type="account_deactivated_login_failures",
user_id=user.id,
source_ip=request.remote_addr
)
log_event(
severity=4,
message=f"Failed login attempt for username: {username}",
event_type="login_failed",
source_ip=request.remote_addr
)
return jsonify({'error': 'Invalid credentials'}), 401
auth_method = "password"
else:
return jsonify({'error': 'Authentication required: provide api_token, client_id/client_secret, or username/password'}), 400
if not user:
return jsonify({'error': 'User not found'}), 404
if not user.is_active:
log_event(
severity=4,
message=f"Login attempt for inactive user: {user.username}",
event_type="login_inactive",
user_id=user.id,
source_ip=request.remote_addr
)
return jsonify({'error': 'Account is inactive'}), 401
# Record successful login
user.record_login_success()
db.session.commit()
# Create access token (convert ID to string for JWT compliance)
expires_delta = timedelta(hours=1)
expiration_time = datetime.now(timezone.utc) + expires_delta
expiration_ts = int(expiration_time.timestamp())
access_token = create_access_token(
identity=str(user.id),
additional_claims={
"client_id": user.username,
"name": user.username,
"expiration": expiration_ts
},
expires_delta=expires_delta
)
log_event(
severity=6,
message=f"Successful login for user: {user.username} via {auth_method}",
event_type="login_success",
user_id=user.id,
source_ip=request.remote_addr
)
return jsonify({
'access_token': access_token,
'token': access_token, # Legacy support
'user': user.to_dict(),
'auth_method': auth_method
})
except Exception as e:
log_event(
severity=3,
message=f"Login error: {str(e)}",
event_type="login_error",
source_ip=request.remote_addr
)
return jsonify({'error': 'Login failed'}), 500
@login_bp.route('/logout', methods=['POST'])
@jwt_required()
def logout():
"""User logout endpoint"""
try:
user_id = get_jwt_identity()
user = User.query.get(user_id)
log_event(
severity=6,
message=f"User logout: {user.username if user else 'unknown'}",
event_type="logout",
user_id=user_id,
source_ip=request.remote_addr
)
return jsonify({'message': 'Logged out successfully'})
except Exception as e:
log_event(
severity=3,
message=f"Logout error: {str(e)}",
event_type="logout_error",
source_ip=request.remote_addr
)
return jsonify({'error': 'Logout failed'}), 500
@login_bp.route('/profile', methods=['GET'])
@jwt_required()
def get_profile():
"""Get current user profile"""
try:
user_id = get_jwt_identity()
user = User.query.get(user_id)
if not user:
return jsonify({'error': 'User not found'}), 404
return jsonify(user.to_dict())
except Exception as e:
log_event(
severity=3,
message=f"Profile error: {str(e)}",
event_type="profile_error",
user_id=get_jwt_identity(),
source_ip=request.remote_addr
)
return jsonify({'error': 'Failed to get profile'}), 500
@login_bp.route('/users', methods=['POST'])
@jwt_required()
def create_user():
"""Create new user (admin only)"""
try:
# Check admin privileges
current_user_id = int(get_jwt_identity())
if not is_admin_user(current_user_id):
log_event(
severity=4,
message=f"Non-admin user {current_user_id} attempted to create user",
event_type="unauthorized_user_creation",
user_id=current_user_id,
source_ip=request.remote_addr
)
return jsonify({'error': 'Admin privileges required'}), 403
data = request.get_json()
if not data:
return jsonify({'error': 'JSON data required'}), 400
username = data.get('username')
email = data.get('email')
password = data.get('password')
# If no password provided, use default and set flag to force change
password_change_required = False
if not password:
password = "bad@pass"
password_change_required = True
if not username or not email:
return jsonify({'error': 'Username and email required'}), 400
# Check if user already exists
existing_user = User.query.filter(
(User.username == username) | (User.email == email)
).first()
if existing_user:
return jsonify({'error': 'User already exists'}), 409
# Create new user
user = User()
user.username = username
user.email = email
user.set_password(password)
user.password_change_required = password_change_required
db.session.add(user)
db.session.commit()
log_event(
severity=6,
message=f"New user created: {username}, password change required: {password_change_required}",
event_type="user_created",
user_id=get_jwt_identity(),
source_ip=request.remote_addr
)
return jsonify(user.to_dict()), 201
except Exception as e:
log_event(
severity=3,
message=f"User creation error: {str(e)}",
event_type="user_creation_error",
user_id=get_jwt_identity(),
source_ip=request.remote_addr
)
return jsonify({'error': 'Failed to create user'}), 500
@login_bp.route('/users/bulk', methods=['POST'])
@jwt_required()
def create_users_bulk():
"""Create multiple users in bulk (admin only)"""
try:
# Check admin privileges
current_user_id = int(get_jwt_identity())
if not is_admin_user(current_user_id):
log_event(
severity=4,
message=f"Non-admin user {current_user_id} attempted to create users in bulk",
event_type="unauthorized_bulk_user_creation",
user_id=current_user_id,
source_ip=request.remote_addr
)
return jsonify({'error': 'Admin privileges required'}), 403
data = request.get_json()
if not data or not isinstance(data, list):
return jsonify({'error': 'JSON array of users required'}), 400
results = {
'success': [],
'failed': []
}
for user_data in data:
try:
if not isinstance(user_data, dict):
results['failed'].append({
'data': user_data,
'error': 'Invalid user data format'
})
continue
username = user_data.get('username')
email = user_data.get('email')
password = user_data.get('password')
# If no username or email, skip this entry
if not username or not email:
results['failed'].append({
'data': user_data,
'error': 'Username and email required'
})
continue
# Check if user already exists
existing_user = User.query.filter(
(User.username == username) | (User.email == email)
).first()
if existing_user:
results['failed'].append({
'data': user_data,
'error': 'User already exists'
})
continue
# If no password provided, use default and set flag to force change
password_change_required = False
if not password:
password = "badpaswword@here"
password_change_required = True
# Create new user
user = User()
user.username = username
user.email = email
user.set_password(password)
user.password_change_required = password_change_required
db.session.add(user)
db.session.flush() # Get the ID without committing
results['success'].append({
'id': user.id,
'username': user.username,
'email': user.email,
'password_change_required': user.password_change_required
})
except Exception as e:
results['failed'].append({
'data': user_data,
'error': str(e)
})
# Commit all successful changes
if results['success']:
db.session.commit()
log_event(
severity=6,
message=f"Bulk user creation: {len(results['success'])} users created, {len(results['failed'])} failed",
event_type="users_created_bulk",
user_id=current_user_id,
source_ip=request.remote_addr
)
else:
db.session.rollback()
return jsonify(results), 201 if results['success'] else 400
except Exception as e:
db.session.rollback()
log_event(
severity=3,
message=f"Bulk user creation error: {str(e)}",
event_type="bulk_user_creation_error",
user_id=int(get_jwt_identity()),
source_ip=request.remote_addr
)
return jsonify({'error': 'Failed to create users', 'details': str(e)}), 500
@login_bp.route('/users/bulk', methods=['GET'])
@jwt_required()
def get_users_bulk():
"""Get all users in a format suitable for bulk creation (admin only)"""
try:
# Check admin privileges
current_user_id = int(get_jwt_identity())
if not is_admin_user(current_user_id):
log_event(
severity=4,
message=f"Non-admin user {current_user_id} attempted to access bulk users list",
event_type="unauthorized_bulk_users_access",
user_id=current_user_id,
source_ip=request.remote_addr
)
return jsonify({'error': 'Admin privileges required'}), 403
# Get all users, including inactive ones if requested
include_inactive = request.args.get('include_inactive', 'false').lower() == 'true'
query = User.query
if not include_inactive:
query = query.filter_by(is_active=True)
users = query.order_by(User.username).all()
# Format users for CSV-compatible output
users_data = []
for user in users:
users_data.append({
'username': user.username,
'email': user.email,
'password': '', # Never include passwords
'require_password_change': user.password_change_required,
'is_active': user.is_active,
'is_admin': user.is_admin or user.username == 'admin'
})
log_event(
severity=6,
message=f"Admin {current_user_id} retrieved bulk users list ({len(users)} users)",
event_type="bulk_users_retrieved",
user_id=current_user_id,
source_ip=request.remote_addr
)
return jsonify({'users': users_data})
except Exception as e:
log_event(
severity=3,
message=f"Bulk users retrieval error: {str(e)}",
event_type="bulk_users_retrieval_error",
user_id=int(get_jwt_identity()),
source_ip=request.remote_addr
)
return jsonify({'error': 'Failed to retrieve users', 'details': str(e)}), 500
@login_bp.route('/generate-api-key', methods=['POST'])
def generate_api_key():
"""Generate API key for user (admin only)"""
try:
data = request.get_json()
if not data:
return jsonify({'error': 'JSON data required'}), 400
# Authenticate the admin making the request
admin_username = data.get('username')
admin_password = data.get('password')
target_user_id = data.get('target_user_id')
key_name = data.get('name')
expiration_days = data.get('expiration_days', 365) # Default 1 year
if not admin_username or not admin_password:
return jsonify({'error': 'Admin username and password required'}), 400
if not target_user_id:
return jsonify({'error': 'target_user_id required'}), 400
# Validate admin credentials
admin_user = User.query.filter_by(username=admin_username).first()
if not admin_user or not admin_user.check_password(admin_password):
log_event(
severity=4,
message=f"Failed API key generation attempt by: {admin_username}",
event_type="api_key_gen_failed",
source_ip=request.remote_addr
)
return jsonify({'error': 'Invalid admin credentials'}), 401
if not admin_user.is_active:
return jsonify({'error': 'Admin account is inactive'}), 401
# For now, assume admin user has username 'admin' (can be enhanced with roles later)
if admin_user.username != 'admin':
log_event(
severity=4,
message=f"Non-admin user {admin_username} attempted to generate API key",
event_type="unauthorized_api_key_gen",
user_id=admin_user.id,
source_ip=request.remote_addr
)
return jsonify({'error': 'Admin privileges required'}), 403
# Validate target user exists
target_user = User.query.get(target_user_id)
if not target_user:
return jsonify({'error': 'Target user not found'}), 404
# Handle no expiration case and ensure type safety
expiration = expiration_days
if expiration_days == 0 or expiration_days is None:
expiration = None
elif not isinstance(expiration_days, int):
expiration = 365 # Default fallback
else:
expiration = expiration_days
# Create API key
api_key = ApiKey.create_key(
user_id=target_user_id,
created_by=admin_user.id,
name=key_name,
expiration_days=expiration
)
db.session.add(api_key)
db.session.commit()
log_event(
severity=6,
message=f"API key generated for user {target_user.username} by admin {admin_username}",
event_type="api_key_generated",
user_id=admin_user.id,
source_ip=request.remote_addr
)
return jsonify({
'api_key': api_key.key,
'name': api_key.name,
'user_id': api_key.user_id,
'expiration_date': api_key.expiration_date.isoformat() if api_key.expiration_date else None,
'expires_in_days': expiration_days
})
except Exception as e:
log_event(
severity=3,
message=f"API key generation error: {str(e)}",
event_type="api_key_gen_error",
source_ip=request.remote_addr
)
return jsonify({'error': 'Failed to generate API key'}), 500
@login_bp.route('/api-key', methods=['POST'])
@jwt_required()
def create_api_key():
"""Create API key with JWT authentication - users can create for themselves, admin can create for anyone"""
try:
data = request.get_json()
if not data:
return jsonify({'error': 'JSON data required'}), 400
current_user_id = int(get_jwt_identity())
target_user_id = data.get('user_id', current_user_id) # Default to self
key_name = data.get('name')
expiration_days = data.get('expiration_days', 365) # Default 1 year
# Check permissions: users can only create keys for themselves, admin can create for anyone
if target_user_id != current_user_id and not is_admin_user(current_user_id):
log_event(
severity=4,
message=f"User {current_user_id} attempted to create API key for user {target_user_id}",
event_type="unauthorized_api_key_creation",
user_id=current_user_id,
source_ip=request.remote_addr
)
return jsonify({'error': 'Can only create API keys for yourself unless you are admin'}), 403
# Validate target user exists
target_user = User.query.get(target_user_id)
if not target_user:
return jsonify({'error': 'Target user not found'}), 404
# Handle no expiration case and ensure type safety
expiration = expiration_days
if expiration_days == 0 or expiration_days is None:
expiration = None
elif not isinstance(expiration_days, int):
expiration = 365 # Default fallback
else:
expiration = expiration_days
# Create API key
api_key = ApiKey.create_key(
user_id=target_user_id,
created_by=current_user_id,
name=key_name,
expiration_days=expiration
)
db.session.add(api_key)
db.session.commit()
log_event(
severity=6,
message=f"API key created for user {target_user.username} by user {current_user_id}",
event_type="api_key_created",
user_id=current_user_id,
source_ip=request.remote_addr
)
return jsonify({
'api_key': api_key.key,
'name': api_key.name,
'user_id': api_key.user_id,
'expiration_date': api_key.expiration_date.isoformat() if api_key.expiration_date else None,
'expires_in_days': expiration
}), 201
except Exception as e:
log_event(
severity=3,
message=f"API key creation error: {str(e)}",
event_type="api_key_creation_error",
user_id=int(get_jwt_identity()),
source_ip=request.remote_addr
)
return jsonify({'error': 'Failed to create API key'}), 500
@login_bp.route('/users', methods=['GET'])
@jwt_required()
def list_users():
"""List users (admin only)"""
try:
# Check admin privileges
current_user_id = int(get_jwt_identity())
if not is_admin_user(current_user_id):
return jsonify({'error': 'Admin privileges required'}), 403
users = User.query.filter_by(is_active=True).all()
users_data = [user.to_dict() for user in users]
log_event(
severity=6,
message=f"Admin {current_user_id} listed {len(users)} users",
event_type="users_listed",
user_id=current_user_id,
source_ip=request.remote_addr
)
return jsonify({'users': users_data})
except Exception as e:
log_event(
severity=3,
message=f"User listing error: {str(e)}",
event_type="user_listing_error",
user_id=int(get_jwt_identity()),
source_ip=request.remote_addr
)
return jsonify({'error': 'Failed to list users'}), 500
@login_bp.route('/api-keys', methods=['GET'])
@jwt_required()
def list_api_keys():
"""List API keys - users see their own keys, admin sees all keys"""
try:
current_user_id = int(get_jwt_identity())
if is_admin_user(current_user_id):
# Admin can see all API keys
api_keys = ApiKey.query.filter_by(is_active=True).all()
message = f"Admin {current_user_id} listed all API keys"
else:
# Regular users can only see their own API keys
api_keys = ApiKey.query.filter_by(user_id=current_user_id, is_active=True).all()
message = f"User {current_user_id} listed their API keys"
keys_data = [key.to_dict() for key in api_keys]
log_event(
severity=6,
message=message,
event_type="api_keys_listed",
user_id=current_user_id,
source_ip=request.remote_addr
)
return jsonify({'api_keys': keys_data})
except Exception as e:
log_event(
severity=3,
message=f"API key listing error: {str(e)}",
event_type="api_key_listing_error",
user_id=int(get_jwt_identity()),
source_ip=request.remote_addr
)
return jsonify({'error': 'Failed to list API keys'}), 500