WebPass / webpass /__init__.py
ag235772's picture
Fixed Socket.IO session setter crash with manage_session=False and pinned versions
adcc4b5
import os
from flask import Flask, session, redirect, url_for, request
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, user_logged_in, current_user
from flask_dance.contrib.google import make_google_blueprint
from flask_socketio import SocketIO
from flask_wtf.csrf import CSRFProtect
from flask_mail import Mail
from flask_migrate import Migrate
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flask_cors import CORS
# Import the Config CLASS
from .config import Config
# Initialize Extensions
db = SQLAlchemy()
socketio = SocketIO()
csrf = CSRFProtect()
login_manager = LoginManager()
mail = Mail()
migrate = Migrate()
limiter = Limiter(key_func=get_remote_address, default_limits=["200 per day", "50 per hour"])
def create_app():
app = Flask(__name__)
# LOAD CONFIGURATION
app.config.from_object(Config)
# --- ENVIRONMENT DETECTION (Cloud vs Local) ---
IS_CLOUD = os.environ.get('SPACE_ID') is not None
if IS_CLOUD:
# Override with SQLite for Hugging Face to prevent MySQL crashes
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///cloud_demo.db'
print("[-] Running in CLOUD MODE (Hugging Face) - Using SQLite")
else:
print("[-] Running in LOCAL MODE - Using Configured Database (MySQL)")
# Initialize Plugins
db.init_app(app)
# --- THE SOCKET FIX IS HERE ---
socketio.init_app(app, async_mode='threading', cors_allowed_origins="*", manage_session=False)
csrf.init_app(app)
login_manager.init_app(app)
mail.init_app(app)
migrate.init_app(app, db)
with app.app_context():
# This will create all tables (User, BiometricDevice, DeadDrop)
# inside fallback.db if they don't exist already.
db.create_all()
print(" [+] Database tables verified/created successfully.")
CORS(app, resources={r"/*": {"origins": "*"}}, supports_credentials=True)
login_manager.login_view = 'auth.login'
from webpass.models import User
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
# 1. SECURITY: RESET BIO
@user_logged_in.connect_via(app)
def on_user_logged_in(sender, user, **extra):
session['bio_verified'] = False
session.permanent = True
# 2. REGISTER BLUEPRINTS
from webpass.routes.auth import auth_bp
from webpass.routes.dashboard import dashboard_bp
from webpass.routes.bio_auth import bio_bp
from webpass.routes.api import api_bp
from webpass.routes.otp import otp_bp
from webpass.routes.share import share_bp
from webpass.routes.stego import stego_bp
from webpass.routes.tools import tools_bp
app.register_blueprint(auth_bp)
app.register_blueprint(dashboard_bp, url_prefix='/dashboard')
app.register_blueprint(bio_bp)
app.register_blueprint(api_bp, url_prefix='/api')
app.register_blueprint(otp_bp)
app.register_blueprint(share_bp)
app.register_blueprint(stego_bp)
app.register_blueprint(tools_bp)
# 3. GOOGLE OAUTH
google_bp = make_google_blueprint(
client_id = app.config["GOOGLE_OAUTH_CLIENT_ID"],
client_secret = app.config["GOOGLE_OAUTH_CLIENT_SECRET"],
scope = ["openid", "https://www.googleapis.com/auth/userinfo.email", "https://www.googleapis.com/auth/userinfo.profile"],
redirect_to = "auth.authorize"
)
google_bp.authorization_url_params["prompt"] = "select_account"
app.register_blueprint(google_bp, url_prefix='/login')
# 4. GLOBAL GATEKEEPER
@app.before_request
def require_biometric_auth():
session.permanent = True
allowed_endpoints = [
'google.login', 'google.authorized', 'auth.login', 'auth.authorize', 'auth.logout',
'static', 'bio.lock_screen', 'bio.mobile_authenticate', 'bio.finalize_login',
'bio.register_begin', 'bio.register_complete', 'bio.auth_begin', 'bio.auth_complete',
'share.view_drop_page', 'share.reveal_drop_api', 'share.create_share', 'share.share_ui'
]
if request.endpoint and request.endpoint not in allowed_endpoints:
if current_user.is_authenticated:
if not session.get('bio_verified'):
return redirect(url_for('bio.lock_screen'))
# 5. SECURITY HEADERS
@app.after_request
def add_security_headers(response):
response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
# SLAMMING THE DOOR: Removed your custom domain since the Vercel Proxy doesn't need it.
# We only allow Hugging Face so your developer portfolio preview still works!
response.headers['Content-Security-Policy'] = "frame-ancestors 'self' https://huggingface.co;"
return response
from webpass.models import BiometricDevice
@app.context_processor
def inject_credential_status():
if current_user.is_authenticated:
try:
device = BiometricDevice.query.filter_by(user_id=current_user.id).first()
return dict(has_credentials=bool(device))
except: pass
return dict(has_credentials=False)
return app