Created
December 31, 2025 19:47
-
-
Save orangewolf/762aaadabdaedd98e14d5f4e0cbe5ee1 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| {% extends "appbuilder/base.html" %} {% block content %} | |
| <div class="container"> | |
| <div | |
| id="loginbox" | |
| style="margin-top: 50px" | |
| class="mainbox col-md-6 col-md-offset-3 col-sm-8 col-sm-offset-2" | |
| > | |
| <div class="panel panel-primary"> | |
| <div class="panel-heading"> | |
| <div class="panel-title">{{ title }}</div> | |
| </div> | |
| <div style="padding-top: 30px" class="panel-body"> | |
| <!-- OAuth Providers Section --> | |
| {% if providers %} | |
| <div style="margin-bottom: 20px"> | |
| <h5>{{ _('Sign in with your account') }}</h5> | |
| {% for provider in providers %} | |
| <a | |
| id="btn-signin-{{provider.name}}" | |
| class="btn btn-primary btn-block" | |
| type="submit" | |
| style="margin-bottom: 10px" | |
| > | |
| {{_('Sign In with ')}}{{ provider.name }} | |
| <i | |
| id="{{provider.name}}" | |
| class="provider-select fa {{provider.icon}} fa-1x" | |
| ></i> | |
| </a> | |
| {% endfor %} | |
| <div style="text-align: center; margin: 20px 0"> | |
| <span style="color: #666">{{ _('or') }}</span> | |
| </div> | |
| </div> | |
| {% endif %} | |
| <!-- Database Login Form Section --> | |
| <div> | |
| <h5>{{ _('Sign in with username and password') }}</h5> | |
| <form class="form" action="" method="post" name="login"> | |
| {{form.hidden_tag()}} | |
| <!-- Username Input --> | |
| <div | |
| class="control-group{% if form.errors.username is defined %} error{% endif %}" | |
| > | |
| <label class="control-label" for="username" | |
| >{{_("Username")}}:</label | |
| > | |
| <div class="controls"> | |
| <div class="input-group"> | |
| <span class="input-group-addon" | |
| ><i class="fa fa-user"></i | |
| ></span> | |
| {{ form.username(size = 80, class = | |
| "form-control", autofocus = true) }} | |
| </div> | |
| {% for error in form.errors.get('username', []) | |
| %} | |
| <span class="help-inline">[{{error}}]</span | |
| ><br /> | |
| {% endfor %} | |
| </div> | |
| </div> | |
| <!-- Password Input --> | |
| <div | |
| class="control-group{% if form.errors.password is defined %} error{% endif %}" | |
| > | |
| <label class="control-label" for="password" | |
| >{{_("Password")}}:</label | |
| > | |
| <div class="controls"> | |
| <div class="input-group"> | |
| <span class="input-group-addon" | |
| ><i class="fa fa-key"></i | |
| ></span> | |
| {{ form.password(size = 80, class = | |
| "form-control") }} | |
| </div> | |
| {% for error in form.errors.get('password', []) | |
| %} | |
| <span class="help-inline">[{{error}}]</span | |
| ><br /> | |
| {% endfor %} | |
| </div> | |
| </div> | |
| <!-- Remember Me Checkbox --> | |
| {% if form.remember_me %} | |
| <div class="control-group"> | |
| <div class="controls"> | |
| <label class="checkbox"> | |
| {{ form.remember_me() }} {{_("Remember | |
| me")}} | |
| </label> | |
| </div> | |
| </div> | |
| {% endif %} | |
| <!-- Login Button --> | |
| <div class="control-group"> | |
| <div class="controls"> | |
| <button | |
| type="submit" | |
| class="btn btn-success btn-block" | |
| > | |
| {{_("Sign In")}} | |
| </button> | |
| </div> | |
| </div> | |
| </form> | |
| </div> | |
| <!-- Registration Link --> | |
| {% if appbuilder.sm.auth_user_registration %} | |
| <div style="margin-top: 20px; text-align: center"> | |
| <a href="{{appbuilder.get_url_for_userregistration}}" | |
| >{{_("Don't have an account? Register here")}}</a | |
| > | |
| </div> | |
| {% endif %} | |
| </div> | |
| </div> | |
| </div> | |
| </div> | |
| <script nonce="{{ baselib.get_nonce() }}"> | |
| {% if providers %} | |
| var baseLoginUrl = "{{appbuilder.get_url_for_login}}"; | |
| var next = "?next=" + "{{request.args.get('next', '') | urlencode}}"; | |
| function signin(provider) { | |
| window.location.href = baseLoginUrl + provider + next; | |
| } | |
| {% for provider in providers %} | |
| document.getElementById("btn-signin-{{provider.name}}") | |
| .addEventListener("click", function () { signin("{{provider.name}}") }) | |
| {% endfor %} | |
| {% endif %} | |
| </script> | |
| {% endblock %} |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| ENABLE_PROXY_FIX = True | |
| from flask_appbuilder.security.manager import (AUTH_DB, AUTH_OAUTH) | |
| AUTH_TYPE = AUTH_OAUTH # This needs to be OAUTH and we'll enable DB auth on the OAUTH action below | |
| AUTH_USER_REGISTRATION = True # allow self-registration (login creates a user) | |
| AUTH_USER_REGISTRATION_ROLE = 'Public' | |
| OAUTH_PROVIDERS = [ | |
| { | |
| 'name': 'google', | |
| 'icon': 'fa-google', | |
| 'token_key': 'access_token', | |
| 'whitelist': [os.getenv("GOOGLE_DOMAIN").strip()], | |
| 'remote_app': { | |
| 'api_base_url': 'https://www.googleapis.com/oauth2/v2/', | |
| 'client_kwargs': { | |
| 'scope': 'email profile' | |
| }, | |
| 'request_token_url': None, | |
| 'access_token_url': 'https://accounts.google.com/o/oauth2/token', | |
| 'authorize_url': 'https://accounts.google.com/o/oauth2/v2/auth', | |
| 'client_id': os.getenv("GOOGLE_CLIENT_ID").strip(), | |
| 'project_id': os.getenv("GOOGLE_PROJECT_ID").strip(), | |
| 'client_secret': os.getenv("GOOGLE_SECRET").strip(), | |
| 'jwks_uri': "https://www.googleapis.com/oauth2/v1/certs", | |
| 'auth_provider_x509_cert_url': "https://www.googleapis.com/oauth2/v1/certs", | |
| "authorize_params": { | |
| "hd": os.getenv("GOOGLE_DOMAIN").strip(), | |
| } | |
| }, | |
| } | |
| ] | |
| AUTH_ROLE_ADMIN = 'Admin' | |
| AUTH_ROLE_PUBLIC = 'Public' | |
| AUTH_USER_REGISTRATION = True | |
| AUTH_USER_REGISTRATION_ROLE = "Gamma" | |
| AUTH_ROLES_MAPPING = { | |
| 'SUPERSET_USERS': ['Gamma'], | |
| 'SUPERSET_ADMIN': ['Admin'] | |
| } | |
| import logging | |
| from typing import Any, List, Optional | |
| from superset.security import SupersetSecurityManager | |
| from flask_appbuilder.security.views import AuthOAuthView | |
| from flask import Flask, flash, g, redirect, request, session, url_for | |
| from flask_login import login_user | |
| from flask_jwt_extended import JWTManager, verify_jwt_in_request | |
| from flask_appbuilder._compat import as_unicode | |
| from flask_appbuilder.security.decorators import no_cache | |
| from flask_appbuilder.security.forms import LoginForm_db | |
| from flask_appbuilder.security.utils import generate_random_string | |
| from flask_appbuilder.utils.base import get_safe_redirect | |
| from flask_appbuilder.views import expose | |
| import jwt | |
| from werkzeug.wrappers import Response as WerkzeugResponse | |
| # CUSTOM_SECURITY_MANAGER = CustomSecurityManager | |
| class CustomAuthOAuthView(AuthOAuthView): | |
| login_template = "loginboth.html" | |
| @expose("/login/", methods=["GET", "POST"]) | |
| @expose("/login/<provider>") | |
| @no_cache | |
| def login(self, provider: Optional[str] = None) -> WerkzeugResponse: | |
| if g.user is not None and g.user.is_authenticated: | |
| return redirect(self.appbuilder.get_url_for_index) | |
| form = LoginForm_db() | |
| if form.validate_on_submit(): | |
| next_url = get_safe_redirect(request.args.get("next", "")) | |
| user = self.appbuilder.sm.auth_user_db( | |
| form.username.data, form.password.data | |
| ) | |
| if not user: | |
| flash(as_unicode(self.invalid_login_message), "warning") | |
| return redirect(self.appbuilder.get_url_for_login_with(next_url)) | |
| login_user(user, remember=False) | |
| return redirect(next_url) | |
| if provider is None: | |
| return self.render_template( | |
| self.login_template, | |
| providers=self.appbuilder.sm.oauth_providers, | |
| form=form, | |
| title=self.title, | |
| appbuilder=self.appbuilder, | |
| ) | |
| random_state = generate_random_string() | |
| state = jwt.encode( | |
| request.args.to_dict(flat=False), random_state, algorithm="HS256" | |
| ) | |
| session["oauth_state"] = random_state | |
| try: | |
| if provider == "twitter": | |
| return self.appbuilder.sm.oauth_remotes[provider].authorize_redirect( | |
| redirect_uri=url_for( | |
| ".oauth_authorized", | |
| provider=provider, | |
| _external=True, | |
| state=state, | |
| ) | |
| ) | |
| else: | |
| return self.appbuilder.sm.oauth_remotes[provider].authorize_redirect( | |
| redirect_uri=url_for( | |
| ".oauth_authorized", provider=provider, _external=True | |
| ), | |
| state=state.decode("ascii") if isinstance(state, bytes) else state, | |
| ) | |
| except Exception as e: | |
| flash(as_unicode(self.invalid_login_message), "warning") | |
| return redirect(self.appbuilder.get_url_for_index) | |
| class CustomSsoSecurityManager(SupersetSecurityManager): | |
| authoauthview = CustomAuthOAuthView | |
| def __init__(self, appbuilder): | |
| super(CustomSsoSecurityManager, self).__init__(appbuilder) | |
| # Add custom template directory to the Jinja2 loader | |
| appbuilder.app.jinja_loader.searchpath.append('/app/docker/pythonpath_dev/templates') | |
| def is_item_public(self, permission_name, view_name): | |
| verify_jwt_in_request(optional=True) # Attempt to parse any existing JWT and fail silently | |
| return super().is_item_public(permission_name, view_name) | |
| def create_jwt_manager(self, app: Flask) -> JWTManager: | |
| def _load_user_jwt(_jwt_header, jwt_data): | |
| user = self.load_user_jwt(_jwt_header, jwt_data) | |
| login_user(user) # sets g.user to jwt provided user | |
| return user | |
| jwt_manager = JWTManager() | |
| jwt_manager.init_app(app) | |
| jwt_manager.user_lookup_loader(_load_user_jwt) | |
| return jwt_manager | |
| def oauth_user_info(self, provider, response=None): | |
| logging.debug('response: {0}'.format(response)) | |
| logging.debug("Oauth2 provider: {0}.".format(provider)) | |
| if provider == "google": | |
| print("one", self.appbuilder.sm.oauth_remotes[provider].get("userinfo")) | |
| me = self.appbuilder.sm.oauth_remotes[provider].get("userinfo") | |
| data = me.json() | |
| print('data is', data) | |
| return { | |
| "username": "google_" + data.get("id", ""), | |
| "first_name": data.get("given_name", ""), | |
| "last_name": data.get("family_name", ""), | |
| "email": data.get("email", ""), | |
| } | |
| CUSTOM_SECURITY_MANAGER = CustomSsoSecurityManager |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment