Files
wfalkowski c177d7981d
Build and publish Docker image / build-and-publish (push) Successful in 2m45s
Refactor backend and add registry build workflow
2026-10-20 16:44:49 +00:00

603 lines
24 KiB
Python

"""Auth API routes for AnchorHost."""
import asyncio
from database_service import database_connection, dict_from_row
from datetime import datetime, timezone
from email_service import get_email_service
from flask import Blueprint, jsonify, redirect, request
import json
import jwt
import os
import requests
from runtime_config import (
APPLE_CLIENT_ID,
APPLE_KEY_ID,
APPLE_PRIVATE_KEY,
APPLE_REDIRECT_URI,
APPLE_TEAM_ID,
GOOGLE_CLIENT_ID,
GOOGLE_CLIENT_SECRET,
GOOGLE_REDIRECT_URI,
connected_clients,
logger,
)
import secrets
import urllib.parse
from user_service import create_user, generate_oauth_state, generate_token, get_user_by_apple_id, get_user_by_email, get_user_by_google_id, verify_oauth_state
from websocket_service import send_message
__all__ = ['apple_callback', 'apple_login', 'get_auth_config', 'google_callback', 'google_login', 'verify_email', 'verify_email_api']
bp = Blueprint("auth", __name__)
@bp.route('/api/auth/google/login', methods=['GET'])
def google_login():
"""Initiate Google OAuth login"""
if not GOOGLE_CLIENT_ID:
return jsonify({'error': 'Google OAuth not configured'}), 500
# Generate state token for CSRF protection
state = generate_oauth_state('google')
# Store state in session or return it to frontend
# For simplicity, we'll include it in the redirect URL
auth_url = (
f"https://accounts.google.com/o/oauth2/v2/auth?"
f"client_id={GOOGLE_CLIENT_ID}&"
f"redirect_uri={urllib.parse.quote(GOOGLE_REDIRECT_URI)}&"
f"response_type=code&"
f"scope=openid email profile&"
f"state={state}&"
f"access_type=offline&"
f"prompt=consent"
)
return jsonify({'auth_url': auth_url, 'state': state})
@bp.route('/api/auth/google/callback', methods=['GET', 'POST'])
def google_callback():
"""Handle Google OAuth callback"""
try:
code = request.args.get('code')
state = request.args.get('state')
if not verify_oauth_state(state, 'google'):
return jsonify({'error': 'Invalid OAuth state'}), 400
if not code:
return jsonify({'error': 'Authorization code not provided'}), 400
# Exchange authorization code for access token
token_url = 'https://oauth2.googleapis.com/token'
token_data = {
'code': code,
'client_id': GOOGLE_CLIENT_ID,
'client_secret': GOOGLE_CLIENT_SECRET,
'redirect_uri': GOOGLE_REDIRECT_URI,
'grant_type': 'authorization_code'
}
token_response = requests.post(token_url, data=token_data)
token_response.raise_for_status()
tokens = token_response.json()
access_token = tokens.get('access_token')
if not access_token:
return jsonify({'error': 'Failed to obtain access token'}), 400
# Get user info from Google
user_info_url = 'https://www.googleapis.com/oauth2/v2/userinfo'
headers = {'Authorization': f'Bearer {access_token}'}
user_info_response = requests.get(user_info_url, headers=headers)
user_info_response.raise_for_status()
google_user = user_info_response.json()
google_id = google_user.get('id')
email = google_user.get('email')
name = google_user.get('name', '')
given_name = google_user.get('given_name', '')
family_name = google_user.get('family_name', '')
picture = google_user.get('picture', '')
verified_email = google_user.get('verified_email', False)
if not google_id or not email:
return jsonify({'error': 'Failed to retrieve user information from Google'}), 400
# Check if user exists by Google ID
user = get_user_by_google_id(google_id)
if not user:
# Check if user exists by email (for linking accounts)
existing_user = get_user_by_email(email)
if existing_user:
# Link Google account to existing user
with database_connection() as conn:
cursor = conn.cursor()
cursor.execute('''
UPDATE users SET google_id = ?, email_verified = ? WHERE id = ?
''', (google_id, True, existing_user['id']))
conn.commit()
user = get_user_by_google_id(google_id)
else:
# Create new user with Google account
# Generate username from email
username_base = email.split('@')[0]
username = username_base
counter = 1
# Ensure username is unique
with database_connection() as conn:
cursor = conn.cursor()
while True:
cursor.execute('SELECT id FROM users WHERE username = ?', (username,))
if not cursor.fetchone():
break
username = f"{username_base}{counter}"
counter += 1
user = create_user(
username=username,
email=email,
password=None, # No password for OAuth users
google_id=google_id,
display_name=name,
first_name=given_name,
last_name=family_name,
email_verified=verified_email
)
if not user:
return jsonify({'error': 'Failed to create or retrieve user'}), 500
# Update last login
with database_connection() as conn:
cursor = conn.cursor()
cursor.execute('''
UPDATE users SET last_login = CURRENT_TIMESTAMP,
login_count = login_count + 1,
email_verified = ?
WHERE id = ?
''', (True, user['id']))
conn.commit()
# Generate JWT token
token = generate_token(user['id'])
# Redirect to frontend with token
frontend_url = os.environ.get('FRONTEND_URL', 'https://anchorhost.eu')
redirect_url = f"{frontend_url}/?token={token}&user={urllib.parse.quote(json.dumps(user))}"
# Redirect to frontend
return redirect(redirect_url)
except requests.exceptions.RequestException as e:
logger.error(f"Google OAuth error: {e}")
return jsonify({'error': 'Failed to authenticate with Google'}), 500
except Exception as e:
logger.error(f"Error in Google OAuth callback: {e}", exc_info=True)
return jsonify({'error': 'An error occurred during authentication'}), 500
@bp.route('/api/auth/apple/login', methods=['GET'])
def apple_login():
"""Initiate Apple OAuth login"""
if not APPLE_CLIENT_ID:
return jsonify({'error': 'Apple OAuth not configured'}), 500
# Generate state token for CSRF protection
nonce = secrets.token_urlsafe(32)
state = generate_oauth_state('apple', nonce)
# Store state in session or return it to frontend
# Apple authorization URL
auth_url = (
f"https://appleid.apple.com/auth/authorize?"
f"client_id={APPLE_CLIENT_ID}&"
f"redirect_uri={urllib.parse.quote(APPLE_REDIRECT_URI)}&"
f"response_type=code id_token&"
f"scope=name email&"
f"response_mode=form_post&"
f"nonce={urllib.parse.quote(nonce)}&"
f"state={state}"
)
return jsonify({'auth_url': auth_url, 'state': state})
@bp.route('/api/auth/apple/callback', methods=['POST'])
def apple_callback():
"""Handle Apple OAuth callback"""
try:
# Apple sends data as form_post (application/x-www-form-urlencoded)
code = request.form.get('code')
id_token = request.form.get('id_token')
state = request.form.get('state')
user_json = request.form.get('user') # Only sent on first login
state_payload = verify_oauth_state(state, 'apple')
if not state_payload or not state_payload.get('nonce'):
return jsonify({'error': 'Invalid OAuth state'}), 400
if not id_token:
return jsonify({'error': 'ID token not provided'}), 400
# Validate Apple's signature, issuer, audience, expiry, and state-bound nonce.
try:
signing_key = jwt.PyJWKClient(
'https://appleid.apple.com/auth/keys',
cache_jwk_set=True,
).get_signing_key_from_jwt(id_token).key
decoded_id_token = jwt.decode(
id_token,
signing_key,
algorithms=['RS256'],
audience=APPLE_CLIENT_ID,
issuer='https://appleid.apple.com',
options={'require': ['exp', 'iat', 'sub', 'nonce']},
)
if not secrets.compare_digest(decoded_id_token['nonce'], state_payload['nonce']):
return jsonify({'error': 'Invalid Apple identity token nonce'}), 400
except Exception as e:
logger.warning(f"Apple identity token validation failed: {e}")
return jsonify({'error': 'Invalid ID token'}), 400
apple_id = decoded_id_token.get('sub')
email = decoded_id_token.get('email')
if not apple_id or not email:
return jsonify({'error': 'Failed to retrieve user information from Apple'}), 400
# Parse user info if provided (name is only sent on first login)
first_name = ""
last_name = ""
display_name = ""
if user_json:
try:
user_data = json.loads(user_json)
name_data = user_data.get('name', {})
first_name = name_data.get('firstName', '')
last_name = name_data.get('lastName', '')
if first_name or last_name:
display_name = f"{first_name} {last_name}".strip()
except:
pass
# Check if user exists by Apple ID
user = get_user_by_apple_id(apple_id)
if not user:
# Check if user exists by email (for linking accounts)
existing_user = get_user_by_email(email)
if existing_user:
# Link Apple account to existing user
with database_connection() as conn:
cursor = conn.cursor()
cursor.execute('''
UPDATE users SET apple_id = ?, email_verified = ? WHERE id = ?
''', (apple_id, True, existing_user['id']))
conn.commit()
user = get_user_by_apple_id(apple_id)
else:
# Create new user with Apple account
# Generate username from email
username_base = email.split('@')[0]
username = username_base
counter = 1
# Ensure username is unique
with database_connection() as conn:
cursor = conn.cursor()
while True:
cursor.execute('SELECT id FROM users WHERE username = ?', (username,))
if not cursor.fetchone():
break
username = f"{username_base}{counter}"
counter += 1
user = create_user(
username=username,
email=email,
password=None, # No password for OAuth users
apple_id=apple_id,
display_name=display_name,
first_name=first_name,
last_name=last_name,
email_verified=True # Apple verified email
)
if not user:
return jsonify({'error': 'Failed to create or retrieve user'}), 500
# Update last login
with database_connection() as conn:
cursor = conn.cursor()
cursor.execute('''
UPDATE users SET last_login = CURRENT_TIMESTAMP,
login_count = login_count + 1,
email_verified = ?
WHERE id = ?
''', (True, user['id']))
conn.commit()
# Generate JWT token
token = generate_token(user['id'])
# Redirect to frontend with token
frontend_url = os.environ.get('FRONTEND_URL', 'https://anchorhost.eu')
redirect_url = f"{frontend_url}/?token={token}&user={urllib.parse.quote(json.dumps(user))}"
# Redirect to frontend
return redirect(redirect_url)
except Exception as e:
logger.error(f"Error in Apple OAuth callback: {e}", exc_info=True)
return jsonify({'error': 'An error occurred during authentication'}), 500
@bp.route('/verify-email', methods=['GET'])
def verify_email():
"""Email verification page"""
token = request.args.get('token')
if not token:
return '''
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Verification Failed</title>
<style>
body { font-family: Arial, sans-serif; background: #0f0f0f; color: #cbcbcd; margin: 0; padding: 40px; }
.container { max-width: 600px; margin: 0 auto; background: #1a1a1a; border-radius: 12px; padding: 40px; box-shadow: 0 4px 6px rgba(0,0,0,0.3); }
h1 { color: #a83a40; margin: 0 0 20px 0; }
p { color: #8a8a8a; line-height: 1.6; }
.btn { display: inline-block; padding: 12px 30px; background: linear-gradient(135deg, #5a7ca7 0%, #4a6385 100%); color: #fff; text-decoration: none; border-radius: 8px; margin-top: 20px; }
</style>
</head>
<body>
<div class="container">
<h1>❌ Verification Failed</h1>
<p>The verification link is invalid or missing.</p>
<a href="/" class="btn">Go to Home</a>
</div>
</body>
</html>
''', 400
try:
with database_connection() as conn:
cursor = conn.cursor()
# Find verification token
cursor.execute('''
SELECT vt.*, u.username, u.email
FROM email_verification_tokens vt
JOIN users u ON vt.user_id = u.id
WHERE vt.token = ?
''', (token,))
token_data = dict_from_row(cursor.fetchone())
if not token_data:
return '''
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Invalid Token</title>
<style>
body { font-family: Arial, sans-serif; background: #0f0f0f; color: #cbcbcd; margin: 0; padding: 40px; }
.container { max-width: 600px; margin: 0 auto; background: #1a1a1a; border-radius: 12px; padding: 40px; box-shadow: 0 4px 6px rgba(0,0,0,0.3); }
h1 { color: #a83a40; margin: 0 0 20px 0; }
p { color: #8a8a8a; line-height: 1.6; }
.btn { display: inline-block; padding: 12px 30px; background: linear-gradient(135deg, #5a7ca7 0%, #4a6385 100%); color: #fff; text-decoration: none; border-radius: 8px; margin-top: 20px; }
</style>
</head>
<body>
<div class="container">
<h1>❌ Invalid Token</h1>
<p>This verification link is invalid or has already been used.</p>
<a href="/" class="btn">Go to Home</a>
</div>
</body>
</html>
''', 404
# Check if token expired
expires_at = datetime.fromisoformat(token_data['expires_at'])
# Fix for offset-naive vs offset-aware comparison
now = datetime.now(timezone.utc) if expires_at.tzinfo else datetime.now()
if now > expires_at:
return '''
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Token Expired</title>
<style>
body { font-family: Arial, sans-serif; background: #0f0f0f; color: #cbcbcd; margin: 0; padding: 40px; }
.container { max-width: 600px; margin: 0 auto; background: #1a1a1a; border-radius: 12px; padding: 40px; box-shadow: 0 4px 6px rgba(0,0,0,0.3); }
h1 { color: #9c6d2f; margin: 0 0 20px 0; }
p { color: #8a8a8a; line-height: 1.6; }
.btn { display: inline-block; padding: 12px 30px; background: linear-gradient(135deg, #5a7ca7 0%, #4a6385 100%); color: #fff; text-decoration: none; border-radius: 8px; margin-top: 20px; }
</style>
</head>
<body>
<div class="container">
<h1>⏰ Token Expired</h1>
<p>This verification link has expired. Please request a new verification email.</p>
<a href="/" class="btn">Go to Home</a>
</div>
</body>
</html>
''', 410
# Verify the email
cursor.execute('''
UPDATE users
SET email_verified = TRUE, email_verification_token = NULL
WHERE id = ?
''', (token_data['user_id'],))
user_updated = cursor.rowcount > 0
# Delete the token
cursor.execute('DELETE FROM email_verification_tokens WHERE token = ?', (token,))
conn.commit()
if user_updated:
# After successful verification, notify the user via WebSocket if they're connected
for client_id, client_info in connected_clients.items():
if (client_info.get('authenticated') and
client_info.get('user_id') == token_data['user_id']):
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
async def send_notification():
await send_message(client_id, {
'type': 'email_verified',
'message': 'Your email has been verified successfully!',
'user_id': token_data['user_id']
})
loop.run_until_complete(send_notification())
loop.close()
except Exception as e:
logger.error(f"Error sending verification notification: {e}")
# Send welcome email
email_svc = get_email_service()
if email_svc:
email_svc.send_welcome_email(token_data['email'], token_data['username'])
logger.info(f"✅ Email verified for user: {token_data['username']}")
return '''
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Email Verified</title>
<style>
body { font-family: Arial, sans-serif; background: #0f0f0f; color: #cbcbcd; margin: 0; padding: 40px; }
.container { max-width: 600px; margin: 0 auto; background: #1a1a1a; border-radius: 12px; padding: 40px; text-align: center; box-shadow: 0 4px 6px rgba(0,0,0,0.3); }
h1 { color: #6b7d55; margin: 0 0 20px 0; font-size: 32px; }
.icon { font-size: 64px; margin-bottom: 20px; }
p { color: #8a8a8a; line-height: 1.6; margin-bottom: 30px; }
.btn { display: inline-block; padding: 16px 40px; background: linear-gradient(135deg, #5a7ca7 0%, #4a6385 100%); color: #fff; text-decoration: none; border-radius: 8px; font-weight: bold; font-size: 16px; }
.btn:hover { background: linear-gradient(135deg, #4a6385 0%, #3a5375 100%); }
</style>
</head>
<body>
<div class="container">
<div class="icon">✅</div>
<h1>Email Verified!</h1>
<p>Your email has been successfully verified. You can now log in and start using AnchorHost!</p>
<a href="/" class="btn">Go to Dashboard</a>
</div>
</body>
</html>
''', 200
except Exception as e:
logger.error(f"Error verifying email: {e}", exc_info=True)
return '''
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Verification Error</title>
<style>
body { font-family: Arial, sans-serif; background: #0f0f0f; color: #cbcbcd; margin: 0; padding: 40px; }
.container { max-width: 600px; margin: 0 auto; background: #1a1a1a; border-radius: 12px; padding: 40px; box-shadow: 0 4px 6px rgba(0,0,0,0.3); }
h1 { color: #a83a40; margin: 0 0 20px 0; }
p { color: #8a8a8a; line-height: 1.6; }
.btn { display: inline-block; padding: 12px 30px; background: linear-gradient(135deg, #5a7ca7 0%, #4a6385 100%); color: #fff; text-decoration: none; border-radius: 8px; margin-top: 20px; }
</style>
</head>
<body>
<div class="container">
<h1>❌ Error</h1>
<p>An error occurred during verification. Please try again or contact support.</p>
<a href="/" class="btn">Go to Home</a>
</div>
</body>
</html>
''', 500
@bp.route('/api/verify-email', methods=['GET', 'POST'])
def verify_email_api():
"""Verify email with token"""
# Handle GET requests by redirecting to the HTML page
if request.method == 'GET':
token = request.args.get('token')
if token:
return redirect(f"/verify-email?token={token}")
return redirect("/verify-email")
try:
data = request.get_json()
token = data.get('token')
if not token:
return jsonify({'success': False, 'error': 'Token required'}), 400
with database_connection() as conn:
cursor = conn.cursor()
# Find token
cursor.execute('SELECT user_id, expires_at FROM email_verification_tokens WHERE token = ?', (token,))
token_data = dict_from_row(cursor.fetchone())
if not token_data:
return jsonify({'success': False, 'error': 'Invalid or expired token'}), 400
expires_at = datetime.fromisoformat(token_data['expires_at'])
# Fix for offset-naive vs offset-aware comparison
now = datetime.now(timezone.utc) if expires_at.tzinfo else datetime.now()
if now > expires_at:
return jsonify({'success': False, 'error': 'Token expired'}), 400
user_id = token_data['user_id']
# Verify user
cursor.execute('UPDATE users SET email_verified = TRUE, email_verification_token = NULL WHERE id = ?', (user_id,))
# Delete token
cursor.execute('DELETE FROM email_verification_tokens WHERE token = ?', (token,))
conn.commit()
# Get user info for welcome email
cursor.execute('SELECT email, username FROM users WHERE id = ?', (user_id,))
user = dict_from_row(cursor.fetchone())
# Send welcome email
if user:
try:
email_svc = get_email_service()
if email_svc:
email_svc.send_welcome_email(user['email'], user['username'])
except Exception as e:
logger.error(f"Failed to send welcome email: {e}")
return jsonify({'success': True, 'message': 'Email verified successfully'}), 200
except Exception as e:
logger.error(f"Error verifying email: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@bp.route('/api/auth/config', methods=['GET'])
def get_auth_config():
"""Get authentication configuration (which OAuth providers are available)"""
def is_valid(val):
return bool(val and str(val).strip() and str(val) not in ['', 'None', 'undefined', 'your.service.identifier', 'your_team_id', 'your_key_id'])
return jsonify({
'google_enabled': is_valid(GOOGLE_CLIENT_ID) and is_valid(GOOGLE_CLIENT_SECRET),
'apple_enabled': is_valid(APPLE_CLIENT_ID) and is_valid(APPLE_TEAM_ID) and is_valid(APPLE_KEY_ID) and is_valid(APPLE_PRIVATE_KEY)
}), 200