mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-09 05:24:49 +09:00
Juan Denis
08d5c4bade
Add full SSO (Google, GitHub, generic OIDC, SAML) support and integrate Alembic-based DB migrations. Introduces OAuthIdentity model, sso_service, SSO API blueprint (authorize/callback/link/unlink/admin/config/test), and settings to enable/configure providers and enforce SSO-only logins. Integrates Flask-Migrate and a MigrationService with API endpoints for status/backup/apply/history and includes Alembic scaffolding (migrations/versions/001_baseline.py). Update app init to register SSO and migrations blueprints and initialize Migrate; replace fragile auto-column hack with MigrationService.prepare/check. Update User model (nullable password_hash, auth_provider, has_password/to_dict fields), add AuditLog SSO actions, add frontend SSO components/pages and a migration UI, and update ROADMAP and requirements. Notes: token encryption uses Fernet derived from SECRET_KEY; migrations may add the new auth_provider column at startup.
329 lines
12 KiB
Python
329 lines
12 KiB
Python
"""SSO / OAuth API blueprint."""
|
|
from datetime import datetime
|
|
from flask import Blueprint, request, jsonify, session
|
|
from flask_jwt_extended import (
|
|
create_access_token, create_refresh_token, jwt_required, get_jwt_identity
|
|
)
|
|
from app import db
|
|
from app.models import User, AuditLog
|
|
from app.models.oauth_identity import OAuthIdentity
|
|
from app.services import sso_service
|
|
from app.services.settings_service import SettingsService
|
|
from app.services.audit_service import AuditService
|
|
|
|
sso_bp = Blueprint('sso', __name__)
|
|
|
|
VALID_PROVIDERS = ('google', 'github', 'oidc', 'saml')
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Public endpoints (login flow)
|
|
# ------------------------------------------------------------------
|
|
|
|
@sso_bp.route('/providers', methods=['GET'])
|
|
def list_providers():
|
|
"""List enabled SSO providers + whether password login is available."""
|
|
return jsonify({
|
|
'providers': sso_service.get_enabled_providers(),
|
|
'password_login_enabled': sso_service.is_password_login_allowed(),
|
|
}), 200
|
|
|
|
|
|
@sso_bp.route('/authorize/<provider>', methods=['GET'])
|
|
def authorize(provider):
|
|
"""Generate OAuth authorize URL (state + PKCE)."""
|
|
if provider not in VALID_PROVIDERS or provider == 'saml':
|
|
return jsonify({'error': f'Invalid OAuth provider: {provider}'}), 400
|
|
|
|
redirect_uri = request.args.get('redirect_uri', '')
|
|
if not redirect_uri:
|
|
return jsonify({'error': 'redirect_uri is required'}), 400
|
|
|
|
try:
|
|
auth_url, state = sso_service.generate_auth_url(provider, redirect_uri)
|
|
return jsonify({'auth_url': auth_url, 'state': state}), 200
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
|
|
@sso_bp.route('/callback/<provider>', methods=['POST'])
|
|
def callback(provider):
|
|
"""Exchange authorization code for tokens, find/create user, return JWT."""
|
|
if provider not in VALID_PROVIDERS or provider == 'saml':
|
|
return jsonify({'error': f'Invalid OAuth provider: {provider}'}), 400
|
|
|
|
data = request.get_json() or {}
|
|
code = data.get('code', '')
|
|
state = data.get('state', '')
|
|
redirect_uri = data.get('redirect_uri', '')
|
|
|
|
if not code or not state:
|
|
return jsonify({'error': 'code and state are required'}), 400
|
|
|
|
try:
|
|
profile = sso_service.handle_oauth_callback(provider, code, state, redirect_uri)
|
|
user, is_new = sso_service.find_or_create_user(provider, profile)
|
|
except ValueError as e:
|
|
AuditLog.log(
|
|
action=AuditLog.ACTION_SSO_LOGIN_FAILED,
|
|
details={'provider': provider, 'error': str(e)},
|
|
ip_address=request.remote_addr,
|
|
)
|
|
db.session.commit()
|
|
return jsonify({'error': str(e)}), 403
|
|
except Exception as e:
|
|
AuditLog.log(
|
|
action=AuditLog.ACTION_SSO_LOGIN_FAILED,
|
|
details={'provider': provider, 'error': str(e)},
|
|
ip_address=request.remote_addr,
|
|
)
|
|
db.session.commit()
|
|
return jsonify({'error': 'SSO authentication failed'}), 500
|
|
|
|
return _complete_sso_login(user, provider, is_new)
|
|
|
|
|
|
@sso_bp.route('/saml/callback', methods=['POST'])
|
|
def saml_callback():
|
|
"""SAML ACS endpoint (form POST from IdP)."""
|
|
try:
|
|
request_data = {
|
|
'https': request.scheme == 'https',
|
|
'http_host': request.host,
|
|
'script_name': request.path,
|
|
'acs_url': request.url,
|
|
'sp_entity_id': request.host_url.rstrip('/'),
|
|
}
|
|
profile = sso_service.handle_saml_callback(request.form, request_data)
|
|
user, is_new = sso_service.find_or_create_user('saml', profile)
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 403
|
|
except Exception as e:
|
|
return jsonify({'error': 'SAML authentication failed'}), 500
|
|
|
|
return _complete_sso_login(user, 'saml', is_new)
|
|
|
|
|
|
@sso_bp.route('/saml/metadata', methods=['GET'])
|
|
def saml_metadata():
|
|
"""Return SP metadata XML."""
|
|
try:
|
|
from onelogin.saml2.auth import OneLogin_Saml2_Auth
|
|
cfg = sso_service.get_provider_config('saml')
|
|
request_data = {
|
|
'https': request.scheme == 'https',
|
|
'http_host': request.host,
|
|
'script_name': request.path,
|
|
'acs_url': request.url_root.rstrip('/') + '/api/v1/sso/saml/callback',
|
|
'sp_entity_id': request.host_url.rstrip('/'),
|
|
}
|
|
settings = sso_service.get_saml_settings(cfg, request_data)
|
|
saml_req = {
|
|
'https': 'on' if request_data.get('https') else 'off',
|
|
'http_host': request_data.get('http_host', ''),
|
|
'script_name': request_data.get('script_name', ''),
|
|
}
|
|
auth = OneLogin_Saml2_Auth(saml_req, settings)
|
|
metadata = auth.get_settings().get_sp_metadata()
|
|
from flask import Response
|
|
return Response(metadata, mimetype='application/xml')
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Authenticated endpoints (account linking)
|
|
# ------------------------------------------------------------------
|
|
|
|
@sso_bp.route('/identities', methods=['GET'])
|
|
@jwt_required()
|
|
def list_identities():
|
|
"""Current user's linked OAuth identities."""
|
|
user_id = get_jwt_identity()
|
|
identities = OAuthIdentity.query.filter_by(user_id=user_id).all()
|
|
return jsonify({'identities': [i.to_dict() for i in identities]}), 200
|
|
|
|
|
|
@sso_bp.route('/link/<provider>', methods=['POST'])
|
|
@jwt_required()
|
|
def link_provider(provider):
|
|
"""Link an OAuth identity to the current user."""
|
|
if provider not in VALID_PROVIDERS:
|
|
return jsonify({'error': f'Invalid provider: {provider}'}), 400
|
|
|
|
user_id = get_jwt_identity()
|
|
data = request.get_json() or {}
|
|
code = data.get('code', '')
|
|
state = data.get('state', '')
|
|
redirect_uri = data.get('redirect_uri', '')
|
|
|
|
if not code or not state:
|
|
return jsonify({'error': 'code and state are required'}), 400
|
|
|
|
try:
|
|
profile = sso_service.handle_oauth_callback(provider, code, state, redirect_uri)
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
# Check if this identity is already linked to another user
|
|
existing = OAuthIdentity.query.filter_by(
|
|
provider=provider,
|
|
provider_user_id=profile['provider_user_id'],
|
|
).first()
|
|
if existing:
|
|
if existing.user_id == user_id:
|
|
return jsonify({'error': 'This identity is already linked to your account'}), 409
|
|
return jsonify({'error': 'This identity is already linked to another account'}), 409
|
|
|
|
identity = sso_service.link_identity(user_id, provider, profile, profile.get('_tokens', {}))
|
|
return jsonify({'identity': identity.to_dict()}), 201
|
|
|
|
|
|
@sso_bp.route('/link/<provider>', methods=['DELETE'])
|
|
@jwt_required()
|
|
def unlink_provider(provider):
|
|
"""Unlink an OAuth identity."""
|
|
if provider not in VALID_PROVIDERS:
|
|
return jsonify({'error': f'Invalid provider: {provider}'}), 400
|
|
|
|
user_id = get_jwt_identity()
|
|
try:
|
|
sso_service.unlink_identity(user_id, provider)
|
|
return jsonify({'message': f'{provider} identity unlinked'}), 200
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Admin endpoints (SSO configuration)
|
|
# ------------------------------------------------------------------
|
|
|
|
@sso_bp.route('/admin/config', methods=['GET'])
|
|
@jwt_required()
|
|
def get_sso_config():
|
|
"""All SSO settings (secrets redacted)."""
|
|
user = User.query.get(get_jwt_identity())
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
config = {}
|
|
for key in SettingsService.DEFAULT_SETTINGS:
|
|
if not key.startswith('sso_'):
|
|
continue
|
|
val = SettingsService.get(key, SettingsService.DEFAULT_SETTINGS[key]['value'])
|
|
# Redact secrets
|
|
if 'secret' in key or 'cert' in key:
|
|
if val and isinstance(val, str) and len(val) > 4:
|
|
val = '****' + val[-4:]
|
|
config[key] = val
|
|
return jsonify({'config': config}), 200
|
|
|
|
|
|
@sso_bp.route('/admin/config/<provider>', methods=['PUT'])
|
|
@jwt_required()
|
|
def update_provider_config(provider):
|
|
"""Update a provider's SSO config."""
|
|
user = User.query.get(get_jwt_identity())
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
if provider not in VALID_PROVIDERS:
|
|
return jsonify({'error': f'Invalid provider: {provider}'}), 400
|
|
|
|
data = request.get_json() or {}
|
|
prefix = f'sso_{provider}_'
|
|
updated = []
|
|
|
|
for key, value in data.items():
|
|
full_key = f'{prefix}{key}'
|
|
if full_key not in SettingsService.DEFAULT_SETTINGS:
|
|
continue
|
|
# Skip unchanged redacted secrets
|
|
if ('secret' in key or 'cert' in key) and isinstance(value, str) and value.startswith('****'):
|
|
continue
|
|
SettingsService.set(full_key, value, user_id=user.id)
|
|
updated.append(key)
|
|
|
|
AuditLog.log(
|
|
action=AuditLog.ACTION_SETTINGS_UPDATE,
|
|
user_id=user.id,
|
|
details={'sso_provider': provider, 'updated_fields': updated},
|
|
)
|
|
db.session.commit()
|
|
return jsonify({'message': f'{provider} SSO config updated', 'updated': updated}), 200
|
|
|
|
|
|
@sso_bp.route('/admin/test/<provider>', methods=['POST'])
|
|
@jwt_required()
|
|
def test_provider(provider):
|
|
"""Test provider connectivity."""
|
|
user = User.query.get(get_jwt_identity())
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
result = sso_service.test_provider_connectivity(provider)
|
|
return jsonify(result), 200 if result['ok'] else 400
|
|
|
|
|
|
@sso_bp.route('/admin/general', methods=['PUT'])
|
|
@jwt_required()
|
|
def update_general_settings():
|
|
"""Update general SSO settings (auto_provision, force_sso, etc.)."""
|
|
user = User.query.get(get_jwt_identity())
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
data = request.get_json() or {}
|
|
general_keys = ['sso_auto_provision', 'sso_default_role', 'sso_force_sso', 'sso_allowed_domains']
|
|
updated = []
|
|
|
|
for key in general_keys:
|
|
if key in data:
|
|
SettingsService.set(key, data[key], user_id=user.id)
|
|
updated.append(key)
|
|
|
|
AuditLog.log(
|
|
action=AuditLog.ACTION_SETTINGS_UPDATE,
|
|
user_id=user.id,
|
|
details={'sso_general': updated},
|
|
)
|
|
db.session.commit()
|
|
return jsonify({'message': 'SSO general settings updated', 'updated': updated}), 200
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helper
|
|
# ------------------------------------------------------------------
|
|
|
|
def _complete_sso_login(user, provider, is_new):
|
|
"""Issue JWT or trigger 2FA for an SSO-authenticated user."""
|
|
# Check 2FA
|
|
if user.totp_enabled:
|
|
temp_token = create_access_token(
|
|
identity=user.id,
|
|
additional_claims={'2fa_pending': True},
|
|
expires_delta=False,
|
|
)
|
|
return jsonify({
|
|
'requires_2fa': True,
|
|
'temp_token': temp_token,
|
|
'message': 'Two-factor authentication required',
|
|
}), 200
|
|
|
|
user.last_login_at = datetime.utcnow()
|
|
user.reset_failed_login()
|
|
db.session.commit()
|
|
|
|
AuditService.log_login(user.id, success=True, details={'provider': provider, 'is_new': is_new})
|
|
db.session.commit()
|
|
|
|
access_token = create_access_token(identity=user.id)
|
|
refresh_token = create_refresh_token(identity=user.id)
|
|
|
|
return jsonify({
|
|
'user': user.to_dict(),
|
|
'access_token': access_token,
|
|
'refresh_token': refresh_token,
|
|
'is_new_user': is_new,
|
|
}), 200
|