"""FastAPI main application entry point.""" from contextlib import asynccontextmanager from pathlib import Path from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from starlette.middleware.sessions import SessionMiddleware from sqlalchemy.orm import Session from .config import get_settings from .database import engine, Base, SessionLocal from .api import api_router from .api.internal import router as internal_router from .web import web_router from .models.user import User, UserRole from .models.tenant import Tenant from .models.endpoint import ApplicationTemplate, DEFAULT_APPLICATION_TEMPLATES from .utils.security import get_password_hash settings = get_settings() # Template directory TEMPLATE_DIR = Path(__file__).parent / "templates" STATIC_DIR = Path(__file__).parent / "static" def init_db(): """Initialize database with tables and default data.""" # Create all tables Base.metadata.create_all(bind=engine) db = SessionLocal() try: # Create default tenant if none exists default_tenant = db.query(Tenant).first() if not default_tenant: default_tenant = Tenant( name="Default", description="Default tenant" ) db.add(default_tenant) db.commit() db.refresh(default_tenant) print("Created default tenant") # Create admin user if none exists admin_user = db.query(User).filter(User.role == UserRole.SUPER_ADMIN).first() if not admin_user: admin_user = User( username=settings.admin_username, email=settings.admin_email, password_hash=get_password_hash(settings.admin_password), role=UserRole.SUPER_ADMIN, tenant_id=None # Super admin has no tenant ) db.add(admin_user) db.commit() print(f"Created admin user: {settings.admin_username}") # Seed application templates existing_templates = db.query(ApplicationTemplate).count() if existing_templates == 0: for template_data in DEFAULT_APPLICATION_TEMPLATES: template = ApplicationTemplate(**template_data) db.add(template) db.commit() print(f"Seeded {len(DEFAULT_APPLICATION_TEMPLATES)} application templates") finally: db.close() @asynccontextmanager async def lifespan(app: FastAPI): """Application lifespan events.""" # Startup print("Starting mGuard VPN Endpoint Server...") init_db() yield # Shutdown print("Shutting down mGuard VPN Endpoint Server...") # Create FastAPI application app = FastAPI( title="mGuard VPN Endpoint Server", description=""" VPN management system for Phoenix Contact mGuard routers. ## Features - Multi-tenant gateway management - Endpoint configuration (IP + Port) - User access control - VPN connection management - Connection logging and auditing """, version="1.0.0", lifespan=lifespan, docs_url="/api/docs", redoc_url="/api/redoc", openapi_url="/api/openapi.json", redirect_slashes=False # Prevent 307 redirects that lose auth headers ) # Session middleware for web UI app.add_middleware( SessionMiddleware, secret_key=settings.secret_key, session_cookie="mguard_session", max_age=86400 * 7, # 7 days ) # Configure CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], # Configure properly in production allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Mount static files app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") # Setup Jinja2 templates templates = Jinja2Templates(directory=str(TEMPLATE_DIR)) app.state.templates = templates # Include API router app.include_router(api_router) # Include Internal API router (for container-to-container communication) app.include_router(internal_router) # Include Web router app.include_router(web_router) @app.get("/") def root(): """Root endpoint with API information.""" return { "name": "mGuard VPN Endpoint Server", "version": "1.0.0", "docs": "/docs", "health": "/health" } @app.get("/health") def health_check(): """Health check endpoint.""" return {"status": "healthy"}