GitHub user Dhruvin4530 edited a discussion: Superset SSO login is not working
after upgrading it to 6.1.0rc1, lending it to username and password page
This is my user_auth file. The SSO is working in v5.
```
"""This file contains user authentication mechanisms for username & password
and also SSO via SAML to Azure AD
"""
import os
import logging
from typing import Optional, Dict, Any, Tuple
from datetime import timedelta
from flask import Flask, session, redirect, request, g
from flask_login import login_user, logout_user
from flask_appbuilder.views import expose
from flask_appbuilder.utils.base import get_safe_redirect
from flask_appbuilder.security.manager import AUTH_DB
from flask_appbuilder.security.views import AuthDBView
from flask_appbuilder.security.sqla.models import User
from flask_appbuilder.const import LOGMSG_WAR_SEC_LOGIN_FAILED
from superset import db, security_manager
# from superset.models.dashboard import Dashboard
from werkzeug.security import check_password_hash
from superset.security import SupersetSecurityManager
from saml2 import (
entity,
BINDING_HTTP_POST,
BINDING_HTTP_REDIRECT,
)
from saml2.client import Saml2Client
from saml2.config import Config as Saml2Config
app = Flask(__name__)
log = logging.getLogger(__name__)
AUTH_TYPE = AUTH_DB
###################################################################################
######################### Superset Environment Variables
##########################
###################################################################################
# Set max age of session to 8 hours
PERMANENT_SESSION_LIFETIME = timedelta(hours=8)
AUTH_USER_REGISTRATION = False
AUTH_USER_REGISTRATION_ROLE = "Public"
AUTH_ROLES_SYNC_AT_LOGIN = False
###################################################################################
########################## Arkose Environment Variables
###########################
###################################################################################
ENVIRONMENT = get_env_variable("ENVIRONMENT", "docker")
WEB_SERVER_HOST_URL = get_env_variable("WEB_SERVER_HOST_URL", "localhost:8088")
AZURE_AD_TENANT_ID = get_env_variable(
"AZURE_AD_TENANT_ID", "f0b24c76-8e72-47e9-b4c2-9022e59e0b5c"
)
"""This is the Arkose specific Tenant ID."""
AZURE_AD_APPLICATION_ID = get_env_variable(
"AZURE_AD_APPLICATION_ID", "3bc6e9ab-0f00-488d-af0e-98adcea3ee89"
)
"""The Azure Active Directory application ID for the Superset application"""
AZURE_AD_GROUP_REQUIRED: str = get_env_variable("AZURE_AD_GROUP_REQUIRED",
False)
"""When SSO is active we can choose to reject users that
did not have a role attribute in the SAML response
"""
ARKOSE_SECURITY_MANAGER = get_env_variable("ARKOSE_SECURITY_MANAGER",
"PASSWORD")
"""Current security manager options are `PASSWORD` or `SSO`"""
# PER APPLICATION configuration settings.
# Each SAML service that you support will have different values here.
metadata_url_for = {
#
https://learn.microsoft.com/en-us/azure/active-directory/develop/active-directory-saml-protocol-reference
#
https://learn.microsoft.com/en-us/azure/active-directory/azuread-dev/azure-ad-federation-metadata#federation-metadata-endpoints
"azure_ad":
f"https://login.microsoftonline.com/{AZURE_AD_TENANT_ID}/FederationMetadata/2007-06/FederationMetadata.xml";,
}
# https://flask-appbuilder.readthedocs.io/en/latest/security.html#role-based
# https://flask-appbuilder.readthedocs.io/en/latest/security.html#permissions
ADMIN_PERMISSIONS = ["Admin"]
SOC_PERMISSIONS = ["Admin", "SOC"]
THR_PERMISSIONS = ["Admin", "THR"]
DATA_PERMISSIONS = ["Admin", "DATA"]
SRE_PERMISSIONS = ["Admin", "SRE"]
ENG_PERMISSIONS = ["Admin", "ENG"]
CSOPS_PERMISSIONS = ["Admin", "CSOPS"]
ENGLEAD_PERMISSIONS = ["Admin", "ENGLEAD"]
PRD_PERMISSIONS = ["Admin", "PRD"]
SRSOC_PERMISSIONS = ["Admin", "SRSOC"]
SE_PERMISSIONS = ["Admin", "SE"]
LVST_PERMISSIONS = ["Admin", "LVST"]
CINFRA_PERMISSIONS = ["Admin", "CINFRA"]
TAM_PERMISSIONS = ["Admin", "TAM"]
DEFAULT_PERMISSIONS = ["Admin", "DEFAULT"]
USER_ROLE_EMAIL_OVERRIDES: Dict[str, Any] = {
"[email protected]": ADMIN_PERMISSIONS,
"[email protected]": ADMIN_PERMISSIONS,
}
# For SSO
#
https://arkoselabs.atlassian.net/wiki/spaces/IT/pages/1784741901/Azure+AD+SSO+Role+Matrix
AZURE_AD_ROLE_MAP = {
"704194f5-733f-445d-9ce8-a8c15ef9f145": {
"name": "IT",
"azure_name": "AAD*_ITAdministrators",
"permissions": ADMIN_PERMISSIONS,
},
"2d98a821-7dcc-426e-8fd7-e253906e8c0b": {
"name": "DATA",
"azure_name": "AAD*_DataEngineers",
"permissions": DATA_PERMISSIONS,
},
"00cd1568-58ba-4fd6-8e64-f5472dc8c6eb": {
"name": "SOC",
"azure_name": "AAD*_SecurityAnalyst",
"permissions": SOC_PERMISSIONS,
},
"ac0ab028-59c6-4793-af04-ecc740970f20": {
"name": "THR",
"azure_name": "AAD*_DataPlatform",
"permissions": THR_PERMISSIONS,
},
"a30217b0-8a22-4532-9863-712cf23450e4": {
"name": "ChirpnContractorsKubernetes",
"azure_name": "AAD*_ChirpnContractorsKubernetes",
"permissions": DEFAULT_PERMISSIONS,
},
"c7f4fc56-c42d-4a52-aeac-80d8576da5bf": {
"name": "HydrolixContractors",
"azure_name": "AAD*_HydrolixContractors",
"permissions": DATA_PERMISSIONS,
},
"c34f3ba4-2f2d-4a33-b08d-d814d3e4679f": {
"name": "SrSecurityAnalyst",
"azure_name": "AAD*_SrSecurityAnalyst",
"permissions": SRSOC_PERMISSIONS,
},
"f7574a3f-b78a-4e9d-b0ab-fa5d8d1d9dff": {
"name": "SRE",
"azure_SSO": "AAD*_SiteReliabilityEngineers",
"permissions": SRE_PERMISSIONS,
},
"d26810ca-6153-4b62-96a5-1343862404ff": {
"name": "ENG",
"azure_SSO": "AAD*_SoftwareEngineers",
"permissions": ENG_PERMISSIONS,
},
"1bee025e-ef81-4410-9402-791b42cbb1da": {
"name": "CSOPS",
"azure_SSO": "AAD*_CSOps",
"permissions": CSOPS_PERMISSIONS,
},
"a94cc806-2137-454e-9118-e6fc499d49dc": {
"name": "ENGLEAD",
"azure_SSO": "AAD*_EngineeringLeadershipTeam",
"permissions": ENGLEAD_PERMISSIONS,
},
"d1f92a95-7fa5-4675-91d3-8a8fc4785098": {
"name": "PRD",
"azure_SSO": "AAD*_Product",
"permissions": PRD_PERMISSIONS,
},
"c34f3ba4-2f2d-4a33-b08d-d814d3e4679f": {
"name": "SRSOC",
"azure_SSO": "AAD*_SrSecurityAnalyst",
"permissions": SRSOC_PERMISSIONS,
},
"a507314b-179d-456e-a330-a71bc412b39d": {
"name": "SE",
"azure_SSO": "AAD*_SrSolutionsEngineer",
"permissions": SE_PERMISSIONS,
},
"fc15d221-a994-43fd-896e-a139b2e6ad57": {
"name": "PDM",
"azure_SSO": "AAD*_PlatformDataManagers",
"permissions": DATA_PERMISSIONS,
},
"bfdd508b-9669-4bf6-8779-28c91d5b9667": {
"name": "LVST",
"azure_SSO": "AAD*_LiveSite",
"permissions": LVST_PERMISSIONS,
},
"11e85f55-c1eb-4b35-b71d-e0f6ad9ee5c6": {
"name": "CINFRA",
"azure_SSO": "AAD*_CloudInfra",
"permissions": CINFRA_PERMISSIONS,
},
"0d232963-10ed-4be2-83aa-384ecd5a240a": {
"name": "TAM",
"azure_SSO": "AAD*_TechnicalAccountManagers",
"permissions": TAM_PERMISSIONS,
},
"47913c36-452c-492c-9c2a-529be43848c4": {
"name": "Program",
"azure_name": "AAD*_Program",
"permissions": ENG_PERMISSIONS,
},
"c362a939-7ac3-4497-b82a-337fefedf295": {
"name": "QualityAssuranceEngineers",
"azure_name": "AAD*_QualityAssuranceEngineers",
"permissions": ENG_PERMISSIONS,
},
"58584972-9ea0-4978-83c9-75e96c17fc94": {
"name": "ChirpnContractorsFalcon",
"azure_name": "AAD*_ChirpnContractorsFalcon",
"permissions": ENG_PERMISSIONS,
},
"cf56d4e3-7db9-4355-8ace-f2969a17e3e3": {
"name": "InfoSec",
"azure_name": "AAD*_InfoSec",
"permissions": ADMIN_PERMISSIONS,
},
"81ea88a1-5c52-488d-9de4-fc097352292b": {
"name": "Networking",
"azure_name": "AAD*_Networking",
"permissions": DEFAULT_PERMISSIONS,
},
"8710a209-ac11-4f70-b3ac-720f08e914c4": {
"name": "ProductDesign",
"azure_name": "AAD*_ProductDesign",
"permissions": DEFAULT_PERMISSIONS,
},
"198479e9-ae48-456e-a53d-dddda18d7140": {
"name": "TechnicalArtist",
"azure_name": "AAD*_TechnicalArtist",
"permissions": DEFAULT_PERMISSIONS,
},
"4bf0ea1a-3ec3-4f1e-af4a-91929190165f": {
"name": "DataEngineersAdmin",
"azure_name": "AAD*_DataEngineersAdmin",
"permissions": ADMIN_PERMISSIONS,
},
"a41ae6d8-1a6e-4259-bc52-cfa6bcf4aa7c": {
"name": "SoftwareEngineersAdmin",
"azure_name": "AAD*_SoftwareEngineersAdmin",
"permissions": ENG_PERMISSIONS,
},
"667bb368-bd94-447a-a523-3a9f3dbf63d2": {
"name": "CSOpsAdmin",
"azure_name": "AAD*_CSOpsAdmin",
"permissions": SOC_PERMISSIONS,
},
"4c91cbbd-6d8f-471c-a2b3-e6f56e79e11e": {
"name": "Detection",
"azure_name": "AAD*_Detection",
"permissions": ENG_PERMISSIONS,
},
"f912ed52-5f46-42a1-9ab0-6daeb6a28875": {
"name": "Enforcement_Services",
"azure_name": "AAD*_Enforcement_Services",
"permissions": ENG_PERMISSIONS,
},
"a860e20d-3533-48d2-85d7-6ea84d136045": {
"name": "Client_Integration",
"azure_name": "AAD*_Client_Integration",
"permissions": ENG_PERMISSIONS,
},
"955a5285-7ce3-4757-92a6-c5672f58f6be": {
"name": "Enforcement",
"azure_name": "AAD*_Enforcement-Team ",
"permissions": ENG_PERMISSIONS,
},
"7211a63f-cce4-49e5-8b04-e4ed38688059": {
"name": "Architecture",
"azure_name": "AAD*_Architecture ",
"permissions": ENG_PERMISSIONS,
},
"6b7ab0a0-a809-4e70-bcd0-8cbf41de8662": {
"name": "Self_Service",
"azure_name": "AAD*_Self_Service",
"permissions": ENG_PERMISSIONS,
},
"67e0b7c1-5845-4a92-b0ad-e7fc96892794": {
"name": "Billing",
"azure_name": "AAD*_Billing",
"permissions": DEFAULT_PERMISSIONS,
},
"9366a944-3795-4b80-9bbe-65ce5c633c73": {
"name": "Hackathon",
"azure_name": "AAD*_Hackathon",
"permissions": DEFAULT_PERMISSIONS,
},
"a7b8f016-bc1a-4d92-b664-c40938e875a3": {
"name": "Accounting",
"azure_name": "AAD*_Accounting",
"permissions": DEFAULT_PERMISSIONS,
},
"04daf140-3c6b-4b1c-88f6-54d104893a1c": {
"name": "AWSMarketplaceAdmin",
"azure_name": "AAD*_AWSMarketplaceAdmin",
"permissions": DEFAULT_PERMISSIONS,
},
"a73e2b8c-de3a-4e09-a9e4-cdb71d5b6b08": {
"name": "CSProduct",
"azure_name": "AAD*_CSProduct",
"permissions": DEFAULT_PERMISSIONS,
},
}
# For Password login
FALLBACK_EMAIL_PERMISSIONS = {
# Admin
"[email protected]": ADMIN_PERMISSIONS,
# Test Users
"[email protected]": ["Admin"],
"[email protected]": DEFAULT_PERMISSIONS,
}
class AuthSAMLView(AuthDBView):
""""""
@expose("/login/", methods=["GET", "POST"])
def login(self):
if g.user is not None and g.user.is_authenticated:
return redirect(self.appbuilder.get_url_for_index)
sm = self.appbuilder.sm
def handle_login():
[redirect_url, req_id, req_info] = sp_initiated_request()
log.info(f"handle_login response: {redirect_url}")
return redirect(redirect_url)
return handle_login()
@expose("/logout/")
def logout(self):
logout_user()
return self.render_template(
"arkose/sso_logout.html",
web_server_host_url=WEB_SERVER_HOST_URL,
)
@expose("/sso/saml/auth/", methods=["POST", "GET"])
def auth(self):
"""This method is responsible for authenticating the returned SSO SAML
response for
users. Users visiting for the first time will have a new user created.
Users not from
an authorised team to access Superset will be rejected.
"""
# See the following about different data that is available from the
request object
# https://tedboy.github.io/flask/generated/generated/flask.Request.html
log.info("SSO called")
sm = self.appbuilder.sm
# Ideally the next_url value should be derived from the original
# target destination of the user
next_url = request.args.get("next", "")
# "JIT provisioning"
authn_response =
sp_initiated_handle_response(request.values["SAMLResponse"])
# auth_identity = authn_response.get_identity()
user_info = authn_response.get_subject()
user_email = None
ava = getattr(authn_response, "ava", {})
for key in [
"emailAddress",
"name",
"http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress";,
]:
if key in ava and ava[key]:
user_email = ava[key][0]
break
# Fallback to NameID if no email claim found
if not user_email:
user_email = getattr(user_info, "text", None)
log.debug(f"authn_response: {authn_response}")
log.info(f"authn_response.id(): {authn_response.id()}")
log.info(f"authn_response.session_id(): {authn_response.session_id()}")
log.debug(f"next_url: {next_url}")
# log.debug(f"authn_response.ava: {authn_response.ava}")
# log.info(f"user_info: {user_info}")
log.info(f"user_email: {user_email}")
# log.info(f"auth_identity: {auth_identity}")
try:
log.info(f"authn_response.session_info():
{authn_response.session_info()}")
except:
pass
try:
log.info(
f"find_encrypt_data:
{authn_response.find_encrypt_data(authn_response)}"
)
except:
pass
group_token = self.get_group_token(authn_response)
log.info(f"group_token: {group_token}")
if group_token is None and AZURE_AD_GROUP_REQUIRED is True:
# Not authorized
return self.render_template(
"arkose/unauthorized.html",
)
if not user_email:
# Not authorized
return self.render_template(
"arkose/unauthorized.html",
)
user = self.auth_user_db(user_email)
if user is None:
log.info(f"Adding new user: {user_email}")
sm.add_user(
authn_response.ava["name"][0],
authn_response.ava["givenName"][0],
authn_response.ava["surname"][0],
authn_response.ava["name"][0],
sm.find_role("Public"),
)
# Fetch the newly created user object
user = sm.find_user(email=authn_response.ava["name"][0])
# Forcing new users to be reauthenticated after making their user.
login_user(user, remember=False)
logout_user()
return redirect(self.appbuilder.get_url_for_login)
update_user_metadata(authn_response)
update_user_roles(user_email, group_token)
login_user(user, remember=False)
return redirect(get_safe_redirect(next_url))
def get_group_token(self, authn_response) -> str:
"""Takes the AuthnResponse from the SAML POST request
Args:
authn_response (AuthnResponse): The response from SAML thing
Returns:
str: The group/role id the authenticating user belongs to
"""
# The following are all response keys the role can be returned from
response_role_keys = [
"role",
"http://schemas.microsoft.com/ws/2008/06/identity/claims/role";,
"groups",
"http://schemas.microsoft.com/ws/2008/06/identity/claims/groups";,
]
for key in response_role_keys:
group_token_list = authn_response.ava.get(key)
if group_token_list is not None:
break
log.debug(f"get_group_token: {group_token_list}")
if not group_token_list:
return None
if group_token_list[0] not in AZURE_AD_ROLE_MAP:
log.info(f"Group token not found in azure mapping:
{group_token_list[0]}")
return group_token_list[0]
def auth_user_db(self, username: str, password: str = None) -> User:
"""Method for authenticating user, auth db style
Args:
username (str): The username or registered email address
password (str, optional): Actually does nothing. Defaults to None.
Returns:
User: The user object matching the provided username
"""
if username is None or username == "":
return None
user = self.appbuilder.sm.find_user(username=username)
if user is None:
user = self.appbuilder.sm.find_user(email=username)
else:
# Balance failure and success
_ = self.appbuilder.sm.find_user(email=username)
# If user is not registered, go away
if user is None:
return None
# If user is not active, go away
if user and (not user.is_active):
self.appbuilder.sm.update_user_auth_stat(user, False)
return None
self.appbuilder.sm.update_user_auth_stat(user, True)
return user
class SAMLSSOSecurityManager(SupersetSecurityManager):
"""Extends the database username and password authentication backend,
however no longer uses the password for authentication. Authentication
is now handled by SSO via Azure Active Directory (the same as other SSO
apps, Slack, aws, etc.)
Use in production"""
def __init__(self, appbuilder):
super(SAMLSSOSecurityManager, self).__init__(appbuilder)
if self.auth_type == AUTH_DB and ARKOSE_SECURITY_MANAGER == "SSO":
self.saml_client = get_saml_client()
self.authdbview = AuthSAMLView
class ModifiedDBSecurityManager(SupersetSecurityManager):
"""Extends the database username and password authentication backend.
Current use case is for non production environment user management"""
user_model = User
authdbview = AuthDBView
def auth_user_db(self, username, password):
"""Method for authenticating user, auth db style
:param username:
The username or registered email address
:param password:
The password, will be tested against hashed password on db
"""
if username is None or username == "":
return None
first_user = self.get_first_user()
user = self.find_user(username=username)
if user is None:
user = self.find_user(email=username)
else:
# Balance failure and success
_ = self.find_user(email=username)
if user is not None:
update_user_roles(user.email)
if user is None or (not user.is_active):
# Balance failure and success
check_password_hash(
"pbkdf2:sha256:150000$Z3t6fmj2$22da622d94a1f8118"
"c0976a03d2f18f680bfff877c9a965db9eedc51bc0be87c",
"password",
)
log.info(LOGMSG_WAR_SEC_LOGIN_FAILED.format(username))
# Balance failure and success
if first_user:
self.noop_user_update(first_user)
return None
elif check_password_hash(user.password, password):
self.update_user_auth_stat(user, True)
return user
else:
self.update_user_auth_stat(user, False)
log.info(LOGMSG_WAR_SEC_LOGIN_FAILED.format(username))
return None
def get_saml_client(host_url: str = None) -> "Saml2Client":
"""Returns a `Saml2Client`.
host_url: the scheme+host seen by the browser (e.g.
https://superset.<environment>.*).
When called inside a request context this should be derived from
request.host_url so
that the SAML ACS URL matches whichever of the three URLs (main, blue,
green) the user
actually used, and Azure posts the callback back to that same URL.
Falls back to WEB_SERVER_HOST_URL when called outside a request context.
"""
acs_base = (host_url or WEB_SERVER_HOST_URL).rstrip("/")
idp_name = "azure_ad"
#
https://github.com/IdentityPython/pysaml2/blob/master/docs/howto/config.rst#configuration-of-pysaml2-entities
settings = {
"entityid": f"spn:{AZURE_AD_APPLICATION_ID}",
"metadata": {
"remote": [
{
"url": metadata_url_for[idp_name],
},
],
},
"key_file": "/app/config/certs/key.pem",
"cert_file": "/app/config/certs/cert.pem",
"service": {
"sp": {
"name": "Superset",
"endpoints": {
"assertion_consumer_service":
[f"{acs_base}/sso/saml/auth/"],
"single_sign_on_service": [
(
f"{acs_base}/sso/saml/auth/",
BINDING_HTTP_REDIRECT,
),
(f"{acs_base}/sso/saml/auth/", BINDING_HTTP_POST),
],
},
# Specifies the constraints on the name identifier to be used to
# represent the requested subject.
# Take a look on
https://learn.microsoft.com/en-us/azure/active-directory/develop/single-sign-on-saml-protocol#nameidpolicy
# to see the NameIdFormat that are supported.
"NameIDFormat":
"urn:oasis:names:tc:SAML:2.0:nameid-format:persistent",
# "NameIDFormat":
"urn:oasis:names:tc:SAML:2.0:nameid-format:emailAddress",
"SPNameQualifier": "",
# Azure supports SHA-256
"signing_algorithm":
"http://www.w3.org/2001/04/xmldsig-more#rsa-sha256";,
# Don't verify that the incoming requests originate from us via
# the built-in cache for authn request ids in pysaml2
"allow_unsolicited": True,
# Don't sign authn requests
"authn_requests_signed": False,
"logout_requests_signed": True,
"want_assertions_signed": True,
"want_response_signed": False,
},
},
"organization": {"display_name": ["Arkose Labs"]},
"contact_person": [
{
"givenname": "Chris",
"surname": "Pidcock",
"phone": "+",
"mail": "[email protected]",
"type": "technical",
},
],
}
sp_config = Saml2Config()
sp_config.load(settings)
sp_config.allow_unknown_attributes = True
saml_client = Saml2Client(config=sp_config)
return saml_client
# https://github.com/IdentityPython/pysaml2/blob/master/docs/howto/config.rst
def sp_initiated_request() -> Tuple:
# Use the host the browser actually used so the ACS URL matches.
# ProxyFix already rewrites request.host_url from X-Forwarded-Host, so
# superset.development.* traffic sees the main URL here, not blue/green.
host_url = request.host_url.rstrip("/")
saml_client = get_saml_client(host_url=host_url)
req_id, req_info = saml_client.prepare_for_authenticate()
redirect_url = metadata_url_for["azure_ad"]
log.info(f"saml_client_req_id: {req_id}")
for key, value in req_info["headers"]:
if key == "Location":
redirect_url = value
break
log.info(f"{redirect_url}")
return redirect_url, req_id, req_info
# https://github.com/IdentityPython/pysaml2/blob/master/docs/howto/config.rst
def sp_initiated_handle_response(data):
host_url = request.host_url.rstrip("/")
saml_client = get_saml_client(host_url=host_url)
authn_response = saml_client.parse_authn_request_response(
xmlstr=data,
binding=entity.BINDING_HTTP_POST,
)
return authn_response
def update_user_metadata(authn_response):
log.info(f"Updating User {authn_response.ava['name'][0]} information")
session = db.session
user = security_manager.find_user(email=authn_response.ava["name"][0])
if user is None:
user = security_manager.find_user(
username=authn_response.ava["name"][0].split("@")[0]
)
user.username = authn_response.ava["name"][0].split("@")[0]
user.first_name = authn_response.ava["givenName"][0]
user.last_name = authn_response.ava["surname"][0]
session.commit()
def update_user_roles(user_email: str, group_token: str = None) -> None:
log.info(f"Updating User {user_email} Roles. SSO group_token:
`{group_token}")
session = db.session
user = security_manager.find_user(email=user_email)
if user.active is False:
user.roles = []
else:
role_list = USER_ROLE_EMAIL_OVERRIDES.get(user_email)
if role_list is None and group_token is not None:
role_dict = AZURE_AD_ROLE_MAP.get(group_token)
if role_dict:
log.info("updating roles from `AZURE_AD_ROLE_MAP`")
role_list = role_dict.get("permissions")
if role_list is None:
log.info("updating roles from `FALLBACK_EMAIL_PERMISSIONS`")
role_list = FALLBACK_EMAIL_PERMISSIONS.get(user_email)
if role_list is None:
log.info("updating roles from `DEFAULT_PERMISSIONS`")
role_list = DEFAULT_PERMISSIONS
if role_list:
user.roles = [
security_manager.find_role(role_name) for role_name in role_list
]
log.info(f"Roles set to: {user_email}={user.roles}")
session.commit()
def enforce_dashboard_certifications(dashboard_id: str) -> None:
"""The idea here was to enforce a certification standard.
The certification standard would ensure specific certification labels
are respected and added only by the owner of a specific label.
What could this look like?
- Teams can commit a config file that defines which dashboards are
authentically certified.
- Owners of a specific certification label should outline what the
label indicated.
"""
log.info(f"Enforcing Dashboard Certifications")
session = db.session
OFFICIAL_CERTIFIED_DASHBOARDS: Dict[str, Any] = {
"Data-Engineering": [
{"id": 0, "name": "best_dashboard_for_real_life"},
],
"soc": [
{"id": 0, "name": "best_dashboard_for_real_life"},
{"id": 1, "name": "slightly_less_best_dashboard_for_real_life"},
],
}
# qry = session.query(Dashboard).filter_by(id="dashboard_id").all()
session.commit()
# FAB_ROLES = {
# "<ROLE NAME>": [
# ["<VIEW/MENU/API NAME>", "PERMISSION NAME"],
# ....
# ],
# ...
# }
FAB_ROLES = {}
def make_session_permanent():
"""Enable maxAge for the cookie `session`"""
session.permanent = True
def FLASK_APP_MUTATOR(app: Flask) -> None:
app.before_request_funcs.setdefault(None, []).append(make_session_permanent)
# Setup SupersetSecurityManager class
if ARKOSE_SECURITY_MANAGER == "SSO":
CUSTOM_SECURITY_MANAGER = SAMLSSOSecurityManager
else:
CUSTOM_SECURITY_MANAGER = ModifiedDBSecurityManager
```
GitHub link: https://github.com/apache/superset/discussions/39948
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
