""" FastAPI Demo App demonstrând sistemul de autentificare ROA2WEB Această aplicație demonstrează integrarea completă a sistemului de autentificare: - Login și logout cu Oracle database - Protected routes cu JWT authentication - Company-specific access control - Permission-based authorization - Rate limiting și security features Funcționează ca: 1. Exemplu de integrare pentru dezvoltatori 2. Tool de testare pentru sistemul de autentificare 3. Demonstrație pentru managementul proiectului Pentru a rula demo-ul: 1. Configurează variabilele de mediu în .env 2. Asigură-te că Oracle database este accesibil 3. Rulează: python demo_app.py 4. Acesează http://localhost:8000/docs pentru Swagger UI """ import asyncio import logging import sys import os from datetime import datetime from typing import List, Optional import uvicorn from fastapi import FastAPI, Depends, HTTPException, Request, status from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse from contextlib import asynccontextmanager # Adaugă calea pentru modulele shared sys.path.append(os.path.join(os.path.dirname(__file__), '..')) # Import modulele de autentificare from .jwt_handler import jwt_handler from .auth_service import auth_service from .models import CurrentUser, LoginRequest, PermissionType from .routes import create_auth_router from .middleware import AuthenticationMiddleware, default_rate_limiter from .dependencies import ( get_current_user, get_optional_user, require_company_access, require_permissions, get_current_company_from_header ) # Import componente shared from database.oracle_pool import oracle_pool from utils.config import shared_config # Configurare logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) @asynccontextmanager async def lifespan(app: FastAPI): """ Lifecycle events pentru demo app """ # Startup logger.info("🚀 Starting ROA2WEB Authentication Demo") try: # Inițializează Oracle pool await oracle_pool.initialize( user=shared_config.oracle_user, password=shared_config.oracle_password, dsn=shared_config.oracle_dsn, min_connections=2, max_connections=5 ) logger.info("✅ Oracle connection pool initialized") # Test database connection async with oracle_pool.get_connection() as conn: with conn.cursor() as cursor: cursor.execute("SELECT 'Database connected successfully' FROM DUAL") result = cursor.fetchone() logger.info(f"✅ Database test: {result[0]}") except Exception as e: logger.error(f"❌ Startup error: {str(e)}") logger.warning("Demo will continue but database features may not work") yield # Shutdown logger.info("🛑 Shutting down ROA2WEB Authentication Demo") try: await oracle_pool.close_pool() logger.info("✅ Oracle connection pool closed") except Exception as e: logger.error(f"❌ Shutdown error: {str(e)}") # Crearea aplicației FastAPI app = FastAPI( title="ROA2WEB Authentication Demo", description=""" Demonstrație completă a sistemului de autentificare ROA2WEB Această aplicație demonstrează: - JWT Authentication cu Oracle Database - Protected routes și company access control - Permission-based authorization - Rate limiting și security features - Integration patterns pentru aplicații ROA2WEB """, version="1.0.0", docs_url="/docs", redoc_url="/redoc", lifespan=lifespan ) # CORS pentru development app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:3000", "http://localhost:5173", "http://localhost:8080"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Authentication middleware app.add_middleware( AuthenticationMiddleware, excluded_paths=["/", "/docs", "/redoc", "/openapi.json", "/health", "/demo", "/auth/login"], rate_limit_paths=["/auth/login"], rate_limiter=default_rate_limiter ) # Include authentication router auth_router = create_auth_router(include_admin_routes=True) app.include_router(auth_router) # ============================================================================= # DEMO ENDPOINTS # ============================================================================= @app.get("/", response_class=HTMLResponse) async def demo_home(): """ Pagina principală cu informații despre demo """ html_content = """ ROA2WEB Authentication Demo

🔐 ROA2WEB Authentication Demo

Status: Demo aplicație ROA2WEB Authentication System
Versiune: 1.0.0
Timp: """ + datetime.now().strftime("%Y-%m-%d %H:%M:%S") + """

📋 Endpoints Disponibile

GET
/docs - Swagger UI pentru testarea API-ului
GET
/health - Health check pentru aplicație și database
POST
/auth/login - Autentificare utilizator cu username/password
GET
/auth/me - Informații utilizator curent (protejat)
GET
/demo/protected - Endpoint protejat simplu
GET
/demo/company/{company_code} - Endpoint cu verificare acces firmă
GET
/demo/admin - Endpoint cu verificare admin permissions

🧪 Cum să testezi

  1. Accesează /docs pentru Swagger UI
  2. Folosește POST /auth/login cu credențiale valide
  3. Copiază access_token din răspuns
  4. Click pe "Authorize" în Swagger UI și introdu: Bearer YOUR_TOKEN
  5. Testează endpoint-urile protejate

🔧 Configurare

Pentru a funcționa complet, demo-ul necesită:

💡 Tip: Pentru dezvoltare rapidă, vezi demo_app.py pentru exemple de integrare a autentificării în aplicațiile tale FastAPI.
""" return HTMLResponse(content=html_content) @app.get("/health") async def health_check(): """ Health check complet pentru demo """ health_status = { "service": "ROA2WEB Authentication Demo", "status": "healthy", "timestamp": datetime.now().isoformat(), "version": "1.0.0" } # Test database connection try: async with oracle_pool.get_connection() as conn: with conn.cursor() as cursor: cursor.execute("SELECT 1 FROM DUAL") cursor.fetchone() health_status["database"] = "connected" except Exception as e: health_status["database"] = f"error: {str(e)}" health_status["status"] = "degraded" # Test JWT handler try: test_token = jwt_handler.create_access_token("healthcheck", ["TEST"]) token_data = jwt_handler.verify_token(test_token) if token_data and token_data.username == "healthcheck": health_status["jwt"] = "functional" else: health_status["jwt"] = "error: token verification failed" health_status["status"] = "degraded" except Exception as e: health_status["jwt"] = f"error: {str(e)}" health_status["status"] = "degraded" # Authentication service status try: cache_stats = auth_service.get_cache_stats() health_status["auth_cache"] = { "total_entries": cache_stats["total_entries"], "cache_hit_ratio": cache_stats["cache_hit_ratio"] } except Exception as e: health_status["auth_cache"] = f"error: {str(e)}" status_code = 200 if health_status["status"] == "healthy" else 503 return JSONResponse(content=health_status, status_code=status_code) @app.get("/demo/public") async def demo_public_endpoint(): """ Endpoint public - nu necesită autentificare """ return { "message": "Acesta este un endpoint public", "authenticated": False, "timestamp": datetime.now().isoformat(), "info": "Acest endpoint poate fi accesat fără autentificare" } @app.get("/demo/optional-auth") async def demo_optional_auth( current_user: Optional[CurrentUser] = Depends(get_optional_user) ): """ Endpoint cu autentificare opțională """ if current_user: return { "message": f"Salut, {current_user.username}!", "authenticated": True, "user": current_user.username, "companies": current_user.companies, "timestamp": datetime.now().isoformat() } else: return { "message": "Acesta este un endpoint cu autentificare opțională", "authenticated": False, "timestamp": datetime.now().isoformat(), "info": "Poți accesa și fără autentificare, dar cu token obții mai multe informații" } @app.get("/demo/protected") async def demo_protected_endpoint( current_user: CurrentUser = Depends(get_current_user) ): """ Endpoint protejat - necesită autentificare """ return { "message": f"Bună ziua, {current_user.username}!", "authenticated": True, "user_info": { "username": current_user.username, "companies": current_user.companies, "permissions": current_user.permissions, "companies_count": len(current_user.companies) }, "timestamp": datetime.now().isoformat(), "info": "Acest endpoint necesită JWT token valid pentru acces" } @app.get("/demo/company/{company_code}") async def demo_company_specific_endpoint( company_code: str, current_user: CurrentUser = Depends(require_company_access("")) # Will be overridden ): """ Endpoint cu verificare acces la firmă specifică Demonstrează cum să verifici dacă utilizatorul are acces la o anumită firmă """ # Verificare manuală pentru demonstrație (în practică folosești dependency) if company_code not in current_user.companies: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"Nu aveți acces la firma {company_code}" ) return { "message": f"Acces permis la firma {company_code}", "company_code": company_code, "user": current_user.username, "user_companies": current_user.companies, "timestamp": datetime.now().isoformat(), "info": "Utilizatorul are acces la această firmă" } @app.get("/demo/admin") async def demo_admin_endpoint( current_user: CurrentUser = Depends(require_permissions([PermissionType.ADMIN])) ): """ Endpoint cu verificare permisiuni admin """ return { "message": f"Bună ziua, admin {current_user.username}!", "admin_info": { "username": current_user.username, "permissions": current_user.permissions, "companies": current_user.companies, "admin_since": datetime.now().isoformat() }, "system_stats": { "total_companies": len(current_user.companies), "demo_version": "1.0.0", "auth_system": "ROA2WEB JWT" }, "timestamp": datetime.now().isoformat(), "info": "Acest endpoint necesită permisiuni de administrator" } @app.get("/demo/reports") async def demo_reports_endpoint( request: Request, current_user: CurrentUser = Depends(require_permissions([PermissionType.REPORTS])) ): """ Endpoint pentru rapoarte - demonstrează integrarea cu header-ul Company """ # Obține company din header (X-Company-Code) sau folosește prima disponibilă company_code = request.headers.get("X-Company-Code") if not company_code: company_code = current_user.companies[0] if current_user.companies else None if not company_code: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Nu s-a specificat codul firmei (X-Company-Code header)" ) if company_code not in current_user.companies: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"Nu aveți acces la rapoartele firmei {company_code}" ) # Simulează generarea unui raport mock_report_data = { "company_code": company_code, "report_type": "demo_report", "generated_by": current_user.username, "generated_at": datetime.now().isoformat(), "data": { "total_invoices": 150, "total_amount": 125000.50, "paid_invoices": 120, "outstanding_amount": 25000.00 } } return { "message": f"Raport generat pentru firma {company_code}", "report": mock_report_data, "user_info": { "username": current_user.username, "permissions": current_user.permissions }, "info": "Acesta este un exemplu de endpoint pentru rapoarte cu verificare company access" } @app.get("/demo/rate-limited") async def demo_rate_limited_endpoint(): """ Endpoint cu rate limiting pentru demonstrație """ return { "message": "Acest endpoint are rate limiting aplicat", "timestamp": datetime.now().isoformat(), "info": "Încercați să faceți mai multe request-uri rapid pentru a vedea rate limiting-ul" } # ============================================================================= # DEMO UTILITIES # ============================================================================= @app.get("/demo/token-info") async def demo_token_info( request: Request, current_user: CurrentUser = Depends(get_current_user) ): """ Endpoint pentru afișarea informațiilor despre token-ul curent """ # Extrage token-ul din header auth_header = request.headers.get("Authorization", "") if auth_header.startswith("Bearer "): token = auth_header[7:] # Decodează token-ul pentru informații (fără verificare pentru demo) payload = jwt_handler.decode_token_payload(token) return { "message": "Informații despre token-ul curent", "token_info": { "user": current_user.username, "companies": current_user.companies, "permissions": current_user.permissions, "token_type": payload.get("type") if payload else "unknown", "issued_at": payload.get("iat") if payload else None, "expires_at": payload.get("exp") if payload else None }, "timestamp": datetime.now().isoformat() } else: return { "error": "Nu s-a găsit token în header-ul Authorization" } # ============================================================================= # MAIN EXECUTION # ============================================================================= def main(): """ Funcția principală pentru rularea demo-ului """ print("🚀 Starting ROA2WEB Authentication Demo") print("📋 Available endpoints:") print(" • http://localhost:8000/ - Demo home page") print(" • http://localhost:8000/docs - Swagger UI") print(" • http://localhost:8000/health - Health check") print(" • http://localhost:8000/demo/* - Demo endpoints") print("") print("💡 Pentru testare completă:") print(" 1. Configurează .env cu credențialele Oracle") print(" 2. Asigură-te că database-ul este accesibil") print(" 3. Folosește /docs pentru testarea interactivă") print("") uvicorn.run( "demo_app:app", host="0.0.0.0", port=8000, reload=True, log_level="info" ) if __name__ == "__main__": main()