Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
File size: 2,999 Bytes
d543fc1 e98a5da d543fc1 | 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 | # pyright: reportGeneralTypeIssues=false, reportAttributeAccessIssue=false
from flask import Flask
from .config import Config, _is_redis_running
from .extensions import db, celery, socketio, limiter, cache
from flask_cors import CORS
def create_app():
app = Flask(__name__)
app.config.from_object(Config)
# Initialize extensions
db.init_app(app)
from .extensions import migrate
migrate.init_app(app, db) # type: ignore
socketio.init_app(app, cors_allowed_origins="*") # type: ignore
app.socketio = socketio # type: ignore
# Configure Caching dynamically based on Redis availability
redis_url = app.config.get('CELERY_BROKER_URL', 'redis://localhost:6379/0')
if _is_redis_running(redis_url):
app.config['CACHE_TYPE'] = 'RedisCache'
app.config['CACHE_REDIS_URL'] = redis_url
else:
app.config['CACHE_TYPE'] = 'SimpleCache'
cache.init_app(app)
# Initialize Firebase Cloud Storage (for logos, PDFs, and file uploads)
from utils.firebase_storage import init_firebase
init_firebase()
# Optional Celery config update
celery.conf.update(
broker_url=redis_url,
result_backend=app.config.get('CELERY_RESULT_BACKEND', 'redis://localhost:6379/0')
)
CORS(app, resources={r"/api/*": {"origins": "*"}}) # type: ignore
with app.app_context():
# Import routes to register blueprints
from .routes import api_bp, auth_bp, scans_bp, vuln_bp, reports_bp
app.register_blueprint(api_bp)
app.register_blueprint(auth_bp, url_prefix='/api/auth')
app.register_blueprint(scans_bp, url_prefix='/api/scans')
app.register_blueprint(vuln_bp, url_prefix='/api/vulnerabilities')
app.register_blueprint(reports_bp, url_prefix='/api/reports')
# Create tables
db.create_all()
# Seed superadmin user (created once, never overwritten)
from .models import User
import os
if not User.query.filter_by(email='superadmin@gmail.com').first():
sa = User(email='superadmin@gmail.com', role='super_admin', org_id=None)
sa.set_password(os.getenv('SUPERADMIN_PASSWORD', 'larshield2025'))
db.session.add(sa)
db.session.commit()
print("[System] Superadmin user seeded (superadmin@gmail.com)")
# Cleanup stale scans that were interrupted during a restart
from .models import Scan
try:
stale_scans = Scan.query.filter(Scan.status.in_(['queued', 'scanning'])).all()
for scan in stale_scans:
scan.status = 'failed'
if stale_scans:
db.session.commit()
print(f"[System] Marked {len(stale_scans)} interrupted scans as failed on startup.")
except Exception as e:
print(f"[System] Failed to clean up stale scans: {e}")
return app
|