| import re |
|
|
| from flask import Blueprint, jsonify, request, abort |
| from flask_jwt_extended import get_jwt_identity |
|
|
| from app import db |
| from auth import jwt_required_with_logging |
| from logging_utils import log_event |
| from models import SamlConfiguration, Tenant, User, ExternalIdentity |
| from saml.exceptions import SamlConfigurationError |
| from saml.service import build_service_urls, parse_idp_metadata |
|
|
| saml_config_bp = Blueprint('saml_config', __name__, url_prefix='/api/saml') |
|
|
|
|
| def _require_admin(): |
| user_id = get_jwt_identity() |
| user = User.query.get(user_id) |
| if not user or not user.is_active or not (user.is_admin or user.username == 'admin'): |
| abort(403, description="Admin privileges required") |
| return user |
|
|
|
|
| def _slugify(value: str) -> str: |
| slug = re.sub(r'[^a-zA-Z0-9-]+', '-', value or '') |
| slug = re.sub(r'-{2,}', '-', slug).strip('-') |
| return slug.lower() |
|
|
|
|
| @saml_config_bp.route('/tenants', methods=['GET']) |
| @jwt_required_with_logging |
| def list_tenants(): |
| _require_admin() |
| tenants = Tenant.query.order_by(Tenant.created_at.desc()).all() |
| return jsonify([tenant.to_dict() for tenant in tenants]) |
|
|
|
|
| @saml_config_bp.route('/tenants', methods=['POST']) |
| @jwt_required_with_logging |
| def create_tenant(): |
| admin = _require_admin() |
| data = request.get_json() or {} |
| name = data.get('name') |
| slug = data.get('slug') or _slugify(name or '') |
| if not name or not slug: |
| abort(400, description="name and slug are required") |
|
|
| if Tenant.query.filter_by(slug=slug).first(): |
| abort(409, description="Tenant slug already exists") |
|
|
| tenant = Tenant( |
| name=name, |
| slug=slug, |
| description=data.get('description'), |
| allow_password_auth=bool(data.get('allow_password_auth', True)), |
| active=bool(data.get('active', True)), |
| ) |
| db.session.add(tenant) |
| db.session.commit() |
|
|
| log_event( |
| severity=6, |
| message=f"Tenant {slug} created by user {admin.username}", |
| event_type="tenant_created", |
| user_id=admin.id, |
| ) |
| return jsonify(tenant.to_dict()), 201 |
|
|
|
|
| @saml_config_bp.route('/tenants/<int:tenant_id>', methods=['PATCH']) |
| @jwt_required_with_logging |
| def update_tenant(tenant_id): |
| admin = _require_admin() |
| tenant = Tenant.query.get_or_404(tenant_id) |
| data = request.get_json() or {} |
|
|
| for field in ('name', 'description', 'active', 'allow_password_auth'): |
| if field in data: |
| setattr(tenant, field, data[field]) |
|
|
| db.session.commit() |
| log_event( |
| severity=6, |
| message=f"Tenant {tenant.slug} updated by user {admin.username}", |
| event_type="tenant_updated", |
| user_id=admin.id, |
| ) |
| return jsonify(tenant.to_dict()) |
|
|
|
|
| @saml_config_bp.route('/tenants/<int:tenant_id>', methods=['DELETE']) |
| @jwt_required_with_logging |
| def delete_tenant(tenant_id): |
| admin = _require_admin() |
| tenant = Tenant.query.get_or_404(tenant_id) |
| tenant_slug = tenant.slug |
| |
| |
| user_count = User.query.filter_by(tenant_id=tenant_id).count() |
| saml_config_count = SamlConfiguration.query.filter_by(tenant_id=tenant_id).count() |
| |
| |
| SamlConfiguration.query.filter_by(tenant_id=tenant_id).delete() |
| |
| |
| ExternalIdentity.query.filter_by(tenant_id=tenant_id).delete() |
| |
| |
| |
| User.query.filter_by(tenant_id=tenant_id).update({'tenant_id': None}) |
| |
| |
| db.session.delete(tenant) |
| db.session.commit() |
| |
| log_event( |
| severity=6, |
| message=f"Tenant {tenant_slug} deleted by user {admin.username} (had {user_count} users, {saml_config_count} SAML configs)", |
| event_type="tenant_deleted", |
| user_id=admin.id, |
| ) |
| |
| return jsonify({ |
| 'message': f'Tenant {tenant_slug} deleted successfully', |
| 'users_affected': user_count, |
| 'saml_configs_deleted': saml_config_count |
| }), 200 |
|
|
|
|
| def _apply_saml_payload(config: SamlConfiguration, data: dict): |
| if 'metadata_xml' in data and data['metadata_xml']: |
| try: |
| parsed = parse_idp_metadata(data['metadata_xml']) |
| except SamlConfigurationError as exc: |
| abort(400, description=str(exc)) |
| config.metadata_xml = data['metadata_xml'] |
| config.idp_entity_id = parsed['entity_id'] |
| config.idp_sso_url = parsed['sso_url'] |
| config.idp_slo_url = parsed.get('slo_url') |
| config.idp_certificate = parsed['certificate'] |
|
|
| for field in ('name', 'enabled', 'auto_provision_users', 'allow_idp_initiated', 'sp_entity_id', 'acs_url', 'slo_url'): |
| if field in data and data[field] is not None: |
| setattr(config, field, data[field]) |
|
|
| attribute_mapping = data.get('attribute_mapping') |
| if attribute_mapping: |
| if not isinstance(attribute_mapping, dict): |
| abort(400, description="attribute_mapping must be a JSON object") |
| config.attribute_mapping = attribute_mapping |
|
|
|
|
| @saml_config_bp.route('/configs', methods=['GET']) |
| @jwt_required_with_logging |
| def list_saml_configs(): |
| _require_admin() |
| configs = SamlConfiguration.query.order_by(SamlConfiguration.created_at.desc()).all() |
| payload = [] |
| for config in configs: |
| data = config.to_dict() |
| data['service_urls'] = build_service_urls(config) |
| payload.append(data) |
| return jsonify(payload) |
|
|
|
|
| @saml_config_bp.route('/configs/<int:config_id>', methods=['GET']) |
| @jwt_required_with_logging |
| def get_saml_config(config_id): |
| _require_admin() |
| config = SamlConfiguration.query.get_or_404(config_id) |
| data = config.to_dict() |
| data['service_urls'] = build_service_urls(config) |
| data['metadata_xml'] = config.metadata_xml |
| data['idp_certificate'] = config.idp_certificate |
| return jsonify(data) |
|
|
|
|
| @saml_config_bp.route('/configs', methods=['POST']) |
| @jwt_required_with_logging |
| def create_saml_config(): |
| admin = _require_admin() |
| data = request.get_json() or {} |
| tenant_id = data.get('tenant_id') |
| metadata_xml = data.get('metadata_xml') |
| name = data.get('name') |
| slug = data.get('slug') or _slugify(name or '') |
|
|
| if not tenant_id or not metadata_xml or not name or not slug: |
| abort(400, description="tenant_id, name, slug, metadata_xml are required") |
|
|
| if SamlConfiguration.query.filter_by(slug=slug).first(): |
| abort(409, description="SAML slug already exists") |
|
|
| tenant = Tenant.query.get_or_404(tenant_id) |
| config = SamlConfiguration( |
| tenant_id=tenant.id, |
| name=name, |
| slug=slug, |
| metadata_xml=metadata_xml, |
| created_by=admin.id, |
| ) |
| _apply_saml_payload(config, data) |
|
|
| urls = build_service_urls(config) |
| config.acs_url = config.acs_url or urls['acs'] |
| config.slo_url = config.slo_url or urls['slo'] |
| config.sp_entity_id = config.sp_entity_id or urls['entity_id'] |
|
|
| db.session.add(config) |
| db.session.commit() |
|
|
| log_event( |
| severity=6, |
| message=f"SAML config {slug} created for tenant {tenant.slug}", |
| event_type="saml_config_created", |
| user_id=admin.id, |
| ) |
|
|
| payload = config.to_dict() |
| payload['service_urls'] = build_service_urls(config) |
| return jsonify(payload), 201 |
|
|
|
|
| @saml_config_bp.route('/configs/<int:config_id>', methods=['PATCH']) |
| @jwt_required_with_logging |
| def update_saml_config(config_id): |
| admin = _require_admin() |
| config = SamlConfiguration.query.get_or_404(config_id) |
| data = request.get_json() or {} |
|
|
| _apply_saml_payload(config, data) |
|
|
| if 'metadata_xml' not in data: |
| |
| urls = build_service_urls(config) |
| if not config.acs_url: |
| config.acs_url = urls['acs'] |
| if not config.slo_url: |
| config.slo_url = urls['slo'] |
| if not config.sp_entity_id: |
| config.sp_entity_id = urls['entity_id'] |
|
|
| db.session.commit() |
| log_event( |
| severity=6, |
| message=f"SAML config {config.slug} updated by user {admin.username}", |
| event_type="saml_config_updated", |
| user_id=admin.id, |
| ) |
| payload = config.to_dict() |
| payload['service_urls'] = build_service_urls(config) |
| return jsonify(payload) |
|
|
