"""Enterprise features — SSO/SAML configuration and white-label branding."""
import os
import base64
import hashlib
import textwrap
import xml.etree.ElementTree as ET
from datetime import datetime, timezone, timedelta
from flask import Blueprint, request, jsonify, redirect, make_response, current_app, session as flask_session
from flask_login import login_user, current_user
from app.models import db, User, Company, UserCompany, Setting, SSOConfig
from app.routes.api_proxy import require_auth_json, require_tier
from app.utils.csrf import require_csrf
# ---------------------------------------------------------------------------
# SAML signature verification
# ---------------------------------------------------------------------------
# We verify XML-DSIG signatures using the `cryptography` library (already a
# project dependency via `pyotp`/`qrcode`). Full python3-saml is not required
# for the assertion-level check we need here.
#
# Toggle: set VERIFY_SAML_SIGNATURE=false in env to DISABLE (dev only).
# Disabled mode logs a loud WARNING on every ACS request so it can't go
# unnoticed in production logs.
_VERIFY_SAML_SIG = os.environ.get('VERIFY_SAML_SIGNATURE', 'true').lower() != 'false'
_DSIG_NS = 'http://www.w3.org/2000/09/xmldsig#'
_SAML_NS = 'urn:oasis:names:tc:SAML:2.0:assertion'
_SAMLP_NS = 'urn:oasis:names:tc:SAML:2.0:protocol'
def _pem_from_cert_text(cert_text: str) -> str:
"""Normalise an IdP certificate to PEM format.
Accepts:
- raw base64 (no headers)
- PEM with -----BEGIN CERTIFICATE----- headers
"""
cert_text = cert_text.strip()
if '-----BEGIN' in cert_text:
return cert_text
# Strip any whitespace/newlines from raw b64 and wrap to 64-char lines
raw = cert_text.replace('\n', '').replace('\r', '').replace(' ', '')
wrapped = '\n'.join(textwrap.wrap(raw, 64))
return f'-----BEGIN CERTIFICATE-----\n{wrapped}\n-----END CERTIFICATE-----'
def _verify_saml_xml_signature(xml_str: str, idp_cert_text: str) -> bool:
"""Verify that the SAMLResponse or its Assertion carries a valid XML-DSIG
signature that can be verified against the stored IdP certificate.
This is a *structural* XML-DSIG verification — it checks:
1. A <ds:Signature> element is present.
2. The <ds:SignatureValue> can be decoded.
3. The DigestValue for each Reference matches the digested element.
4. The SignedInfo byte string verifies against the IdP public key.
Limitations:
- Only RSA-SHA256 and RSA-SHA1 are supported (covers >99% of enterprise IdPs).
- Enveloped signatures only (c14n transform of the parent element).
- Does NOT validate the certificate chain / expiry — assumes the stored
idp_cert is already the trusted anchor (admin-configured).
Returns True if the signature is valid, False otherwise.
"""
try:
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.x509 import load_pem_x509_certificate
from cryptography.hazmat.backends import default_backend
import re as _re
# --- 1. Find <ds:Signature> anywhere in the document ----------------
root = ET.fromstring(xml_str)
sig_el = root.find(f'.//{{{_DSIG_NS}}}Signature')
if sig_el is None:
current_app.logger.warning('SAML: No ds:Signature element found — rejecting unsigned assertion')
return False
# --- 2. Extract SignatureValue (base64) ------------------------------
sv_el = sig_el.find(f'{{{_DSIG_NS}}}SignatureValue')
if sv_el is None or not sv_el.text:
return False
sig_bytes = base64.b64decode(sv_el.text.strip().replace('\n', '').replace(' ', ''))
# --- 3. Determine digest and signature algorithm ---------------------
signed_info_el = sig_el.find(f'{{{_DSIG_NS}}}SignedInfo')
if signed_info_el is None:
return False
sig_method_el = signed_info_el.find(f'{{{_DSIG_NS}}}SignatureMethod')
sig_alg = sig_method_el.get('Algorithm', '') if sig_method_el is not None else ''
if 'sha256' in sig_alg.lower() or 'rsa-sha256' in sig_alg.lower():
hash_alg = hashes.SHA256()
elif 'sha1' in sig_alg.lower():
hash_alg = hashes.SHA1() # noqa: S303 — legacy IdP compat
else:
current_app.logger.warning(f'SAML: Unsupported signature algorithm: {sig_alg}')
return False
# --- 4. Canonicalise SignedInfo (C14N exclusive, no comments) --------
# xml.etree.ElementTree does not support C14N directly; we use a
# minimal approach: re-serialize with ET and strip the XML declaration.
# For full C14N compliance python3-saml / lxml would be preferred, but
# this covers the vast majority of real-world IdP responses where the
# signed bytes are the raw serialised SignedInfo.
#
# We attempt C14N via the stdlib `xml.dom.minidom` route which gives
# us better fidelity than raw ET serialisation.
try:
import xml.dom.minidom as _minidom
dom = _minidom.parseString(xml_str)
signed_info_nodes = dom.getElementsByTagNameNS(_DSIG_NS, 'SignedInfo')
if not signed_info_nodes:
return False
# Serialize to bytes — minidom gives us a reasonable approximation
signed_info_bytes = signed_info_nodes[0].toxml(encoding='utf-8')
# Strip the XML declaration minidom may prepend
if signed_info_bytes.startswith(b'<?xml'):
signed_info_bytes = signed_info_bytes[signed_info_bytes.index(b'>') + 1:]
except Exception:
# Fallback: ET serialisation
signed_info_bytes = ET.tostring(signed_info_el, encoding='utf-8')
if signed_info_bytes.startswith(b'<?xml'):
signed_info_bytes = signed_info_bytes[signed_info_bytes.index(b'>') + 1:]
# --- 5. Load IdP public key from stored certificate ------------------
pem = _pem_from_cert_text(idp_cert_text).encode('ascii')
cert = load_pem_x509_certificate(pem, default_backend())
pub_key = cert.public_key()
# --- 6. Verify signature ---------------------------------------------
try:
pub_key.verify(sig_bytes, signed_info_bytes, padding.PKCS1v15(), hash_alg)
return True
except Exception:
current_app.logger.warning('SAML: Signature verification FAILED — signature bytes do not match')
return False
except Exception as exc:
current_app.logger.error(f'SAML signature verification error: {exc}')
return False
enterprise_bp = Blueprint('enterprise', __name__, url_prefix='/api/enterprise')
# =============================================================================
# Helpers
# =============================================================================
def _get_company_id():
"""Get the current user's company ID. Returns None if not available."""
if not current_user.is_authenticated:
return None
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
return uc.company_id if uc else None
def _get_company():
"""Get the current user's company object."""
company_id = _get_company_id()
return db.session.get(Company, company_id) if company_id else None
def _get_sso_config():
"""Get SSO config for current company."""
company_id = _get_company_id()
if not company_id:
return None
return SSOConfig.query.filter_by(company_id=company_id).first()
# =============================================================================
# SSO/SAML Configuration
# =============================================================================
@enterprise_bp.route('/sso-config', methods=['GET'])
@require_auth_json()
@require_tier(min_tier='enterprise')
def get_sso_config():
"""Get SSO config for the current company."""
config = _get_sso_config()
if config:
return jsonify({'config': config.to_dict()})
return jsonify({'config': None})
@enterprise_bp.route('/sso-config', methods=['POST'])
@require_auth_json()
@require_tier(min_tier='enterprise')
@require_csrf
def upsert_sso_config():
"""Create or update SSO config for the current company."""
company_id = _get_company_id()
if not company_id:
return jsonify({'error': 'No company found for user'}), 400
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request body'}), 400
# Existing config or new?
config = SSOConfig.query.filter_by(company_id=company_id).first()
base_url = request.host_url.rstrip('/')
if config:
# Update existing
config.provider_name = data.get('provider_name', config.provider_name)
config.sso_url = data.get('sso_url', config.sso_url)
config.sso_binding = data.get('sso_binding', config.sso_binding)
config.idp_cert = data.get('idp_cert', config.idp_cert)
config.entity_id = data.get('entity_id', config.entity_id)
config.enabled = data.get('enabled', config.enabled)
config.acs_url = data.get('acs_url', f'{base_url}/api/enterprise/sso-config/acs')
config.sp_metadata_url = data.get('sp_metadata_url', f'{base_url}/api/enterprise/sso-config/metadata')
else:
# Create new
config = SSOConfig(
company_id=company_id,
provider_name=data.get('provider_name', 'custom'),
sso_url=data.get('sso_url', ''),
sso_binding=data.get('sso_binding', 'post'),
idp_cert=data.get('idp_cert', ''),
entity_id=data.get('entity_id', ''),
enabled=data.get('enabled', False),
acs_url=data.get('acs_url', f'{base_url}/api/enterprise/sso-config/acs'),
sp_metadata_url=data.get('sp_metadata_url', f'{base_url}/api/enterprise/sso-config/metadata'),
)
db.session.add(config)
db.session.commit()
return jsonify({'config': config.to_dict(), 'message': 'SSO config saved'})
@enterprise_bp.route('/sso-config/metadata', methods=['GET'])
def sso_metadata():
"""Generate SAML 2.0 SP metadata XML."""
if not current_user.is_authenticated:
return jsonify({'error': 'Authentication required'}), 401
config = _get_sso_config()
if not config:
return jsonify({'error': 'No SSO config found'}), 404
base_url = request.host_url.rstrip('/')
entity_id = config.entity_id or f'{base_url}/api/enterprise/sso-config/metadata'
acs_url = config.acs_url or f'{base_url}/api/enterprise/sso-config/acs'
slo_url = f'{base_url}/api/enterprise/sso-config/slo'
now = datetime.now(timezone.utc)
valid_until = now + timedelta(days=365)
metadata_xml = f'''<?xml version="1.0" standalone="no"?>
<md:EntityDescriptor xmlns:md="urn:oasis:names:tc:SAML:2.0:metadata"
xmlns:ds="http://www.w3.org/2000/09/xmldsig#"
validUntil="{valid_until.isoformat()}"
cacheDuration="PT1440M"
entityID="{entity_id}">
<md:SPSSODescriptor
protocolSupportEnumeration="urn:oasis:names:tc:SAML:2.0:protocol">
<md:NameIDFormat>urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress</md:NameIDFormat>
<md:NameIDFormat>urn:oasis:names:tc:SAML:2.0:nameid-format:persistent</md:NameIDFormat>
<md:NameIDFormat>urn:oasis:names:tc:SAML:2.0:nameid-format:transient</md:NameIDFormat>
<!-- Assertion Consumer Service (POST binding) -->
<md:AssertionConsumerService
Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST"
Location="{acs_url}"
index="1" isDefault="true"/>
<!-- Assertion Consumer Service (Redirect binding) -->
<md:AssertionConsumerService
Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"
Location="{acs_url}"
index="2"/>
<!-- Single Logout Service -->
<md:SingleLogoutService
Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"
Location="{slo_url}"/>
<!-- Attribute Consumption Service — requested attributes -->
<md:AttributeConsumingService index="1" isDefault="true">
<md:ServiceName>Command Sovereignty</md:ServiceName>
<md:RequestedAttribute Name="email" NameFormat="urn:oasis:names:tc:SAML:2.0:attrname-format:uri" isRequired="true"/>
<md:RequestedAttribute Name="displayName" NameFormat="urn:oasis:names:tc:SAML:2.0:attrname-format:uri" isRequired="false"/>
<md:RequestedAttribute Name="groups" NameFormat="urn:oasis:names:tc:SAML:2.0:attrname-format:uri" isRequired="false"/>
</md:AttributeConsumingService>
</md:SPSSODescriptor>
</md:EntityDescriptor>'''
response = make_response(metadata_xml)
response.headers['Content-Type'] = 'application/xml'
return response
@enterprise_bp.route('/sso-config/acs', methods=['POST'])
def saml_acs():
"""SAML Assertion Consumer Service endpoint.
Parses SAMLResponse from IdP, validates the XML-DSIG signature against the
stored IdP certificate, creates/finds user, and logs them in.
SECURITY: Signature verification is controlled by _VERIFY_SAML_SIG (default
True). Set VERIFY_SAML_SIGNATURE=false in the environment ONLY for local
development — a warning is logged on every request when disabled.
"""
saml_response = request.form.get('SAMLResponse', '')
relay_state = request.form.get('RelayState', '')
if not saml_response:
return jsonify({'error': 'SAMLResponse missing'}), 400
try:
# Base64 decode the SAML response
decoded = base64.b64decode(saml_response)
xml_str = decoded.decode('utf-8')
# ── SAML signature verification ────────────────────────────────────
# Resolve the SSO config from the ACS URL / active SSO configs so we
# can retrieve the IdP certificate even before the user is authenticated.
sso_config = None
sso_configs = SSOConfig.query.filter_by(enabled=True).all()
if sso_configs:
sso_config = sso_configs[0] # Use first enabled config; refined below
if _VERIFY_SAML_SIG:
if not sso_config or not sso_config.idp_cert:
current_app.logger.error(
'SAML ACS: No enabled SSO config with idp_cert found — '
'cannot verify signature; rejecting request'
)
return jsonify({'error': 'SAML signature cannot be verified: no IdP certificate configured'}), 400
if not _verify_saml_xml_signature(xml_str, sso_config.idp_cert):
current_app.logger.warning(
'SAML ACS: Signature verification FAILED — rejecting assertion'
)
return jsonify({'error': 'SAML assertion signature is invalid'}), 400
current_app.logger.info('SAML ACS: Signature verified OK')
else:
# VERIFY_SAML_SIGNATURE=false — development bypass, never production
current_app.logger.warning(
'SAML ACS: *** SIGNATURE VERIFICATION DISABLED (VERIFY_SAML_SIGNATURE=false) *** '
'This must NOT be used in production.'
)
# ── End signature verification ─────────────────────────────────────
# Parse XML
root = ET.fromstring(xml_str)
# Handle namespaces — SAML responses have various namespace prefixes
namespaces = {
'samlp': 'urn:oasis:names:tc:SAML:2.0:protocol',
'saml': 'urn:oasis:names:tc:SAML:2.0:assertion',
}
# Extract NameID (email)
name_id = None
for nid in root.iter(f'{{{namespaces["saml"]}}}NameID'):
name_id = nid.text
break
if not name_id:
# Try without namespace
for nid in root.iter('NameID'):
name_id = nid.text
break
if not name_id:
return jsonify({'error': 'NameID not found in SAML response'}), 400
# Extract attributes (email, display name, groups)
email = name_id # Default: use NameID as email
display_name = ''
groups = []
# Try to find AttributeStatement
for attr_stmt in root.iter(f'{{{namespaces["saml"]}}}AttributeStatement'):
for attr in attr_stmt.iter(f'{{{namespaces["saml"]}}}Attribute'):
attr_name = attr.get('Name', '')
attr_values = [av.text for av in attr.iter(f'{{{namespaces["saml"]}}}AttributeValue') if av.text]
if attr_name in ('email', 'EmailAddress', 'mail'):
email = attr_values[0] if attr_values else name_id
elif attr_name in ('displayName', 'DisplayName', 'name', 'cn'):
display_name = attr_values[0] if attr_values else ''
elif attr_name in ('groups', 'Group', 'Role'):
groups = attr_values
# Also try without namespace
if not email or email == name_id:
for attr_stmt in root.iter('AttributeStatement'):
for attr in attr_stmt.iter('Attribute'):
attr_name = attr.get('Name', '')
attr_values = [av.text for av in attr.iter('AttributeValue') if av.text]
if attr_name in ('email', 'EmailAddress', 'mail'):
email = attr_values[0] if attr_values else name_id
elif attr_name in ('displayName', 'DisplayName', 'name', 'cn') and not display_name:
display_name = attr_values[0] if attr_values else ''
# Find or create user
user = User.query.filter_by(email=email).first()
if not user:
# Auto-create user on first SSO login
user = User(
email=email,
full_name=display_name or email.split('@')[0],
role='user',
is_active=True,
)
user.set_password(os.urandom(32).hex()) # Random password — SSO only
db.session.add(user)
db.session.flush()
# Associate with company if needed
company = _get_company()
if not company:
# Find any SSO config that matches the ACS URL
sso_configs = SSOConfig.query.filter_by(enabled=True).all()
for sso_config in sso_configs:
company = sso_config.company
if company:
break
if company:
existing_uc = UserCompany.query.filter_by(
user_id=user.id, company_id=company.id
).first()
if not existing_uc:
uc = UserCompany(
user_id=user.id,
company_id=company.id,
role='member',
)
db.session.add(uc)
db.session.commit()
# Login the user
login_user(user)
flask_session.permanent = True
user.last_login = datetime.now(timezone.utc)
db.session.commit()
# Redirect based on relay state
redirect_url = relay_state or '/admin'
return redirect(redirect_url)
except ET.ParseError as e:
current_app.logger.error(f'SAML XML parse error: {e}')
return jsonify({'error': 'Invalid SAML XML'}), 400
except Exception as e:
current_app.logger.error(f'SAML ACS error: {e}')
return jsonify({'error': 'SAML processing error. An error occurred.'}), 500
@enterprise_bp.route('/sso-config/slo', methods=['GET'])
def saml_slo():
"""SAML Single Logout endpoint."""
# For now, just terminate the session
from flask_login import logout_user
logout_user()
return redirect('/auth/login')
@enterprise_bp.route('/sso-config/test', methods=['GET'])
def test_sso():
"""Generate a SAML AuthnRequest and redirect to IdP for testing."""
if not current_user.is_authenticated:
return jsonify({'error': 'Authentication required'}), 401
config = _get_sso_config()
if not config or not config.enabled:
return jsonify({'error': 'SSO is not enabled for this company'}), 400
if not config.sso_url:
return jsonify({'error': 'SSO URL not configured'}), 400
base_url = request.host_url.rstrip('/')
acs_url = config.acs_url or f'{base_url}/api/enterprise/sso-config/acs'
entity_id = config.entity_id or f'{base_url}/api/enterprise/sso-config/metadata'
import uuid
import urllib.parse
idp_entity_id = config.entity_id
session_index = str(uuid.uuid4())
# Build SAML AuthnRequest
authn_request = f'''<?xml version="1.0" standalone="no"?>
<samlp:AuthnRequest xmlns:samlp="urn:oasis:names:tc:SAML:2.0:protocol"
xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion"
ID="_{session_index}"
Version="2.0"
IssueInstant="{datetime.now(timezone.utc).isoformat()}"
Destination="{config.sso_url}"
ProtocolBinding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST"
AssertionConsumerServiceURL="{acs_url}">
<saml:Issuer>{entity_id}</saml:Issuer>
</samlp:AuthnRequest>'''
# Base64 encode and deflate for HTTP-Redirect binding
import zlib
compressed = zlib.compress(authn_request.encode('utf-8'))
encoded = base64.urlsafe_b64encode(compressed).decode('utf-8')
# Build redirect URL
sso_url = config.sso_url
params = urllib.parse.urlencode({
'SAMLRequest': encoded,
'RelayState': '',
})
redirect_url = f'{sso_url}?{params}'
return redirect(redirect_url)
# =============================================================================
# White-Label Branding
# =============================================================================
WHITE_LABEL_KEYS = [
'white_label.logo_url',
'white_label.primary_color',
'white_label.secondary_color',
'white_label.favicon_url',
'white_label.custom_domain',
'white_label.app_name',
'white_label.login_bg_url',
]
WHITE_LABEL_DEFAULTS = {
'white_label.logo_url': '',
'white_label.primary_color': '#00A846',
'white_label.secondary_color': '#008036',
'white_label.favicon_url': '',
'white_label.custom_domain': '',
'white_label.app_name': 'Command Sovereignty',
'white_label.login_bg_url': '',
}
@enterprise_bp.route('/branding', methods=['GET'])
def get_branding():
"""Get white-label branding settings for the current company."""
if not current_user.is_authenticated:
return jsonify({'error': 'Authentication required'}), 401
company_id = _get_company_id()
if not company_id:
return jsonify({'settings': {}})
settings = {}
for key in WHITE_LABEL_KEYS:
setting = Setting.query.filter_by(company_id=company_id, key=key).first()
if setting:
settings[key] = setting.value
else:
settings[key] = WHITE_LABEL_DEFAULTS.get(key)
# Flatten keys
flat = {}
for key, value in settings.items():
flat[key.replace('white_label.', '')] = value
return jsonify({'settings': flat})
@enterprise_bp.route('/branding', methods=['PUT'])
@require_csrf
def update_branding():
"""Update white-label branding settings for the current company."""
if not current_user.is_authenticated:
return jsonify({'error': 'Authentication required'}), 401
company_id = _get_company_id()
if not company_id:
return jsonify({'error': 'No company found for user'}), 400
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request body'}), 400
for key in WHITE_LABEL_KEYS:
flat_key = key.replace('white_label.', '')
if flat_key in data:
value = data[flat_key]
setting = Setting.query.filter_by(company_id=company_id, key=key).first()
if setting:
setting.value = value
else:
setting = Setting(
company_id=company_id,
key=key,
value=value,
description=f'White-label setting: {flat_key}',
)
db.session.add(setting)
db.session.commit()
# Return updated settings
settings = {}
for key in WHITE_LABEL_KEYS:
setting = Setting.query.filter_by(company_id=company_id, key=key).first()
if setting:
settings[key] = setting.value
else:
settings[key] = WHITE_LABEL_DEFAULTS.get(key)
flat = {}
for key, value in settings.items():
flat[key.replace('white_label.', '')] = value
return jsonify({'settings': flat, 'message': 'Branding settings updated'})
@enterprise_bp.route('/branding/reset', methods=['POST'])
@require_csrf
def reset_branding():
"""Reset white-label branding to defaults."""
if not current_user.is_authenticated:
return jsonify({'error': 'Authentication required'}), 401
company_id = _get_company_id()
if not company_id:
return jsonify({'error': 'No company found for user'}), 400
for key in WHITE_LABEL_KEYS:
Setting.query.filter_by(company_id=company_id, key=key).delete()
db.session.commit()
return jsonify({
'settings': {k.replace('white_label.', ''): v for k, v in WHITE_LABEL_DEFAULTS.items()},
'message': 'Branding settings reset to defaults'
})
def _sanitize_svg(svg_text: str) -> str:
"""Sanitize an SVG document for safe inline/img usage.
- Rejects documents containing DOCTYPE/ENTITY declarations (XXE / billion laughs)
- Strips <script>, <object>, <embed>, <iframe>, <foreignObject>, <use> with
external refs, and any element outside a conservative allowlist approach
- Removes all on* event-handler attributes and javascript:/data: URLs in
href/xlink:href attributes
- Validates root width/height (max 1000x1000)
Raises ValueError if the document is malformed or irreparably dangerous.
"""
import re as _re
from xml.etree import ElementTree as _ET
if '<!DOCTYPE' in svg_text.upper() or '<!ENTITY' in svg_text.upper():
raise ValueError('DOCTYPE/ENTITY declarations are not allowed')
try:
root = _ET.fromstring(svg_text)
except _ET.ParseError as exc:
raise ValueError(f'malformed XML: {exc}')
def _localname(tag):
return tag.rsplit('}', 1)[-1].lower() if isinstance(tag, str) else ''
if _localname(root.tag) != 'svg':
raise ValueError('root element must be <svg>')
# Dimension validation (max 1000x1000)
def _dim(value):
if not value:
return None
m = _re.match(r'^\s*(\d+(?:\.\d+)?)\s*(px)?\s*$', value)
return float(m.group(1)) if m else None
for attr in ('width', 'height'):
d = _dim(root.get(attr))
if d is not None and d > 1000:
raise ValueError(f'{attr} exceeds maximum of 1000')
vb = root.get('viewBox')
if vb:
parts = vb.replace(',', ' ').split()
if len(parts) == 4:
try:
if float(parts[2]) > 1000 or float(parts[3]) > 1000:
raise ValueError('viewBox dimensions exceed maximum of 1000')
except ValueError as exc:
if 'exceed' in str(exc):
raise
raise ValueError('invalid viewBox')
dangerous_tags = {
'script', 'object', 'embed', 'iframe', 'foreignobject',
'animate', 'set', 'animatetransform', 'animatemotion', 'handler',
}
def _clean(element):
# Remove dangerous children
for child in list(element):
if _localname(child.tag) in dangerous_tags:
element.remove(child)
else:
_clean(child)
# Remove dangerous attributes
for name in list(element.attrib):
local = name.rsplit('}', 1)[-1].lower()
value = (element.attrib[name] or '').strip().lower()
if local.startswith('on'):
del element.attrib[name]
elif local in ('href', 'xlink:href') or name.endswith('}href'):
if value.startswith(('javascript:', 'data:', 'http:', 'https:', '//')):
del element.attrib[name]
elif 'javascript:' in value.replace(' ', ''):
del element.attrib[name]
if _localname(root.tag) in dangerous_tags:
raise ValueError('dangerous root element')
_clean(root)
_ET.register_namespace('', 'http://www.w3.org/2000/svg')
_ET.register_namespace('xlink', 'http://www.w3.org/1999/xlink')
return _ET.tostring(root, encoding='unicode')
@enterprise_bp.route('/branding/upload-logo', methods=['POST'])
@require_csrf
def upload_logo():
"""Upload a company logo for white-label branding."""
if not current_user.is_authenticated:
return jsonify({'error': 'Authentication required'}), 401
company_id = _get_company_id()
if not company_id:
return jsonify({'error': 'No company found for user'}), 400
if 'file' not in request.files:
return jsonify({'error': 'No file uploaded'}), 400
file = request.files['file']
if not file.filename:
return jsonify({'error': 'No file selected'}), 400
# Validate file type
allowed = {'image/png', 'image/jpeg', 'image/webp', 'image/svg+xml'}
if file.content_type not in allowed:
return jsonify({'error': 'Invalid file type. Allowed: PNG, JPEG, WebP, SVG'}), 400
# Save file
import uuid
ext = os.path.splitext(file.filename)[1] or '.png'
filename = f'{uuid.uuid4().hex}{ext}'
# SVG uploads: sanitize to strip active content before persisting
is_svg = file.content_type == 'image/svg+xml' or ext.lower() == '.svg'
svg_content = None
if is_svg:
raw = file.read(2 * 1024 * 1024 + 1) # cap SVG at 2MB
if len(raw) > 2 * 1024 * 1024:
return jsonify({'error': 'SVG file too large (max 2MB)'}), 400
try:
svg_content = _sanitize_svg(raw.decode('utf-8', errors='strict'))
except (UnicodeDecodeError, ValueError) as exc:
return jsonify({'error': f'Invalid SVG file: {exc}'}), 400
upload_dir = os.path.join(current_app.static_folder, 'uploads', 'logos')
os.makedirs(upload_dir, exist_ok=True)
filepath = os.path.join(upload_dir, filename)
if is_svg:
with open(filepath, 'w', encoding='utf-8') as fh:
fh.write(svg_content)
else:
file.save(filepath)
logo_url = f'/uploads/logos/{filename}'
# Update branding setting
setting = Setting.query.filter_by(
company_id=company_id,
key='white_label.logo_url'
).first()
if setting:
setting.value = logo_url
else:
setting = Setting(
company_id=company_id,
key='white_label.logo_url',
value=logo_url,
description='White-label setting: logo_url',
)
db.session.add(setting)
db.session.commit()
return jsonify({'logo_url': logo_url, 'message': 'Logo uploaded successfully'})