File size: 5,283 Bytes
136c0f7 adcc4b5 136c0f7 adcc4b5 188b6eb adcc4b5 136c0f7 2145e85 3f87767 2145e85 136c0f7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | 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 |