haproxy-openmanager/backend/main.py

from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import logging
import json
import os
import redis
import asyncio
from datetime import datetime, timedelta

_version_info = {"version": "1.3.0", "releaseName": "Bulk Import Change Detection + Multi-Select Delete", "releaseDate": "2026-04-14"}
for _vpath in ["/app/version.json", os.path.join(os.path.dirname(__file__), "..", "version.json")]:
    try:
        with open(_vpath) as _vf:
            _version_info = json.load(_vf)
        break
    except (FileNotFoundError, json.JSONDecodeError):
        continue

# Version: 2026-04-02 - Dark Mode + UI Improvements v1.2.0
# Version: 2026-04-01 - ACME Auto SSL v1.1.0
# Version: 2025-10-20 - Agent script fixes deployed

# Import configurations and database

from config import CORS_ORIGINS, REDIS_URL, LOG_LEVEL
from database.connection import redis_client, get_database_connection, close_database_connection, init_database_pool, close_database_pool
from database.migrations import run_all_migrations

# Import routers
from routers import frontend_router, backend_router, cluster_router, dashboard_router, agent_router, waf_router, ssl_router, auth_router, user_router
from routers.health import router as health_router
from routers.config import router as config_router
from routers.security import router as security_router
from routers.configuration import router as configuration_router
from routers.maintenance import router as maintenance_router
from routers.dashboard_stats import router as dashboard_stats_router
from routers.settings import router as settings_router
from routers.letsencrypt import router as letsencrypt_router

# Production logging configuration
from utils.logging_config import setup_production_logging
from middleware.error_handler import (
    RequestLoggingMiddleware, PerformanceMonitoringMiddleware, 
    GlobalExceptionHandler, get_error_statistics
)
from middleware.activity_logger import log_activity_middleware

# Setup structured logging
logger = setup_production_logging(LOG_LEVEL)

# Initialize FastAPI app with detailed API documentation
app = FastAPI(
    title="HAProxy Open Manager API", 
    version=_version_info["version"],
    description="""
# HAProxy Open Manager - Enterprise Multi-Cluster Management API

🚀 **Production-ready HAProxy management system** with agent-pull architecture for managing multiple HAProxy clusters remotely.

## Architecture Overview

This system uses an **agent-pull architecture**:
- Changes are NOT pushed from backend to agents
- Agent services periodically poll the backend for tasks
- Agents execute tasks on their local HAProxy servers
- Agents report status and sync configuration back to backend

## Getting Started

### Authentication
All API endpoints (except `/api/auth/login` and agent endpoints) require JWT authentication:

```bash
# 1. Login to get access token
curl -X POST "{BASE_URL}/api/auth/login" \\
  -H "Content-Type: application/json" \\
  -d '{"username": "admin", "password": "admin123"}'

# Response: {"access_token": "eyJ...", "token_type": "bearer"}

# 2. Use token in subsequent requests
curl -X GET "{BASE_URL}/api/clusters" \\
  -H "Authorization: Bearer eyJ..."
```

**Note:** Replace `{BASE_URL}` with your actual deployment URL (e.g., `https://haproxy-manager.company.com`)

## Workflow

### Initial Setup (From Scratch)

1. **Create Agent Pool**
   - Pools group agents logically
   - POST `/api/clusters/pools` - Create a new pool

2. **Create HAProxy Cluster**
   - Select the pool created above
   - POST `/api/clusters` - Create cluster associated with pool

3. **Generate Agent Installation Script**
   - Select pool when creating agent
   - POST `/api/agents` - Generate platform-specific installation script
   - Agent automatically gets associated with cluster via pool

4. **Install Agent on Remote HAProxy Server**
   - Download and run the generated script on target server
   - Agent service starts and begins polling backend
   - Agent logs written to `/var/log/haproxy-agent/agent.log`

5. **Configure HAProxy via UI**
   - Define backends, frontends, WAF rules, SSL certificates
   - Changes are stored in backend database
   - Agents periodically pull changes and apply to local haproxy.cfg
   - HAProxy service is automatically reloaded

### Multi-Cluster Management

Most operations are **cluster-scoped**:
- Select cluster in UI (top bar cluster selector)
- All operations apply to selected cluster only
- Exception: SSL Management can work across all clusters

### Agent Version Management

- Agent versions stored in backend database
- Agents check for updates on each poll cycle
- If new version available, agent downloads and self-updates
- Update agent scripts via UI: `Agent Management > Update Script`

## Key Features

- ✅ Multi-cluster management
- ✅ Agent-pull architecture (no push from backend)
- ✅ Backend & Frontend management
- ✅ WAF rule management
- ✅ SSL certificate management
- ✅ Bulk configuration import
- ✅ Real-time agent health monitoring
- ✅ Configuration version history
- ✅ Role-based access control (RBAC)
- ✅ Audit logging

## Deployment

- Platform: OpenShift
- CI/CD: Azure DevOps Pipeline
- Architecture: Agent-Pull based distributed management
""",
    contact={
        "name": "HAProxy Open Manager Team",
        "email": "support@example.com",
    },
    license_info={
        "name": "Enterprise License",
    },
    docs_url="/api/docs",
    redoc_url="/api/redoc",
    openapi_url="/api/openapi.json",
    openapi_tags=[
        {
            "name": "Authentication",
            "description": "User authentication and JWT token management. All protected endpoints require a valid JWT token obtained from `/api/auth/login`."
        },
        {
            "name": "Users",
            "description": "User management operations including CRUD operations, role assignments, and user activity tracking."
        },
        {
            "name": "clusters",
            "description": "HAProxy cluster management. Clusters represent groups of HAProxy instances that share configuration. Each cluster is associated with an agent pool."
        },
        {
            "name": "pools",
            "description": "Agent pool management. Pools are logical groups that connect agents to clusters. Create pools first, then create clusters referencing them."
        },
        {
            "name": "agents",
            "description": "Agent management and monitoring. Agents run on HAProxy servers and pull configuration from backend. Includes installation script generation, version management, and health monitoring."
        },
        {
            "name": "frontends",
            "description": "HAProxy frontend management. Frontends define how HAProxy receives incoming traffic (bind addresses, ports, SSL/TLS configuration)."
        },
        {
            "name": "backends",
            "description": "HAProxy backend and server management. Backends define pools of servers that handle requests, including load balancing algorithms and health checks."
        },
        {
            "name": "waf",
            "description": "Web Application Firewall (WAF) rule management. Define security rules to protect web applications from common attacks."
        },
        {
            "name": "ssl",
            "description": "SSL/TLS certificate management. Upload, manage, and deploy SSL certificates across clusters. Supports per-cluster and all-cluster deployment."
        },
        {
            "name": "configuration",
            "description": "Configuration management including bulk import, version history, and configuration preview. Import existing HAProxy configs or view generated configurations."
        },
        {
            "name": "dashboard",
            "description": "Real-time statistics and monitoring dashboard. View cluster health, agent status, traffic metrics, and system overview."
        },
        {
            "name": "security",
            "description": "Security and audit features including activity logs, access control, and system security settings."
        },
        {
            "name": "maintenance",
            "description": "System maintenance operations including database cleanup, soft-delete recovery, and system health checks."
        },
        {
            "name": "health",
            "description": "Health check endpoints for monitoring system status, database connectivity, and Redis status."
        }
    ]
)

# Background task for agent status monitoring
async def monitor_agent_status():
    """Background task to monitor agent status and mark offline agents"""
    while True:
        try:
            conn = await get_database_connection()
            
            # Mark agents as offline if they haven't sent heartbeat in last 120 seconds (2 minutes)
            threshold_time = datetime.utcnow() - timedelta(seconds=120)
            
            result = await conn.execute("""
                UPDATE agents 
                SET status = 'offline', updated_at = CURRENT_TIMESTAMP
                WHERE status = 'online' 
                AND (last_seen IS NULL OR last_seen < $1)
            """, threshold_time)
            
            await close_database_connection(conn)
            
            # Log if any agents were marked offline
            if result and hasattr(result, 'split') and len(result.split()) > 1:
                count = result.split()[1]
                if int(count) > 0:
                    logger.info(f"Marked {count} agents as offline due to missing heartbeats")
            
        except Exception as e:
            logger.error(f"Error in agent status monitoring: {e}")
        
        # Wait 30 seconds before next check
        await asyncio.sleep(30)

# Background task for ACME certificate auto-renewal
async def check_letsencrypt_renewals():
    """Background task to auto-renew expiring ACME certificates."""
    await asyncio.sleep(120)
    while True:
        conn = None
        try:
            conn = await get_database_connection()
            table_exists = await conn.fetchval("""
                SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'letsencrypt_orders')
            """)
            if not table_exists:
                await close_database_connection(conn)
                conn = None
                await asyncio.sleep(3600)
                continue

            auto_renew_setting = await conn.fetchval(
                "SELECT value FROM system_settings WHERE key = 'acme.auto_renew_enabled'"
            )
            auto_renew_enabled = False
            if auto_renew_setting:
                try:
                    val = json.loads(auto_renew_setting) if isinstance(auto_renew_setting, str) else auto_renew_setting
                    auto_renew_enabled = val in (True, 'true', 'True')
                except (json.JSONDecodeError, TypeError):
                    pass

            if not auto_renew_enabled:
                await close_database_connection(conn)
                conn = None
                await asyncio.sleep(3600)
                continue

            renew_days_raw = await conn.fetchval(
                "SELECT value FROM system_settings WHERE key = 'acme.renew_before_days'"
            )
            renew_days = 30
            if renew_days_raw:
                try:
                    renew_days = int(json.loads(renew_days_raw) if isinstance(renew_days_raw, str) else renew_days_raw)
                except (ValueError, TypeError, json.JSONDecodeError):
                    pass

            expiring_certs = await conn.fetch("""
                SELECT id, name, primary_domain, letsencrypt_order_id
                FROM ssl_certificates
                WHERE source = 'letsencrypt' AND auto_renew = TRUE AND is_active = TRUE
                AND expiry_date IS NOT NULL
                AND expiry_date <= (CURRENT_DATE + ($1 || ' days')::INTERVAL)
            """, str(renew_days))

            await close_database_connection(conn)
            conn = None

            from routers.letsencrypt import _complete_certificate
            from services.acme_service import acme_service as acme_svc

            # --- Phase 1: Create new orders for expiring certificates ---
            for cert in expiring_certs:
                try:
                    order_id = cert['letsencrypt_order_id']
                    if not order_id:
                        continue

                    conn2 = await get_database_connection()
                    order = None
                    domains = None
                    cluster_ids = None
                    skip = False
                    try:
                        order = await conn2.fetchrow(
                            "SELECT account_id, domains, cluster_ids FROM letsencrypt_orders WHERE id = $1",
                            order_id
                        )
                        if order:
                            domains = json.loads(order['domains']) if isinstance(order['domains'], str) else order['domains']
                            cluster_ids = json.loads(order['cluster_ids']) if isinstance(order['cluster_ids'], str) else order['cluster_ids']
                            existing = await conn2.fetchrow("""
                                SELECT id FROM letsencrypt_orders
                                WHERE domains::text = $1::text
                                AND status NOT IN ('valid', 'invalid', 'cancelled')
                                AND created_at > NOW() - INTERVAL '48 hours'
                                LIMIT 1
                            """, json.dumps(domains))
                            if existing:
                                logger.info(f"ACME RENEWAL: Skipping cert {cert['id']} - order {existing['id']} already in progress")
                                skip = True
                    finally:
                        await close_database_connection(conn2)

                    if not order or skip:
                        continue

                    new_order = await acme_svc.create_order(order['account_id'], domains, cluster_ids)
                    await acme_svc.respond_to_challenges(new_order['order_id'])
                    logger.info(f"ACME RENEWAL: Initiated renewal order {new_order['order_id']} for cert {cert['id']} ({cert['name']})")
                except Exception as cert_err:
                    logger.error(f"ACME RENEWAL ERROR: Failed to initiate renewal for cert {cert['id']}: {cert_err}")

            # --- Phase 2: Poll and complete in-progress orders ---
            conn_poll = await get_database_connection()
            try:
                in_progress = await conn_poll.fetch("""
                    SELECT id, status FROM letsencrypt_orders
                    WHERE (
                        status IN ('pending', 'processing', 'ready')
                        OR (status = 'valid' AND ssl_certificate_id IS NULL)
                    )
                    AND created_at > NOW() - INTERVAL '7 days'
                    ORDER BY created_at
                """)
            finally:
                await close_database_connection(conn_poll)

            for order_row in in_progress:
                oid = order_row['id']
                try:
                    status_info = await acme_svc.check_order_status(oid)
                    current_status = status_info.get('status', order_row['status'])

                    if current_status == 'ready':
                        await acme_svc.finalize_order(oid)
                        status_info = await acme_svc.check_order_status(oid)
                        current_status = status_info.get('status')

                    if current_status == 'valid' and status_info.get('certificate_url'):
                        result = await _complete_certificate(oid)
                        logger.info(f"ACME RENEWAL: Completed order {oid} - {result.get('message', '')}")
                    elif current_status == 'invalid':
                        logger.warning(f"ACME RENEWAL: Order {oid} is invalid, skipping")
                    elif current_status in ('pending', 'processing'):
                        logger.info(f"ACME RENEWAL: Order {oid} still {current_status}, will retry next cycle")
                except Exception as poll_err:
                    logger.error(f"ACME RENEWAL: Failed to poll/complete order {oid}: {poll_err}")

            # --- Phase 3: Warn about stuck orders ---
            conn3 = await get_database_connection()
            try:
                stuck_orders = await conn3.fetch("""
                    SELECT id, status, domains, created_at FROM letsencrypt_orders
                    WHERE status IN ('pending', 'processing')
                    AND created_at < NOW() - INTERVAL '24 hours'
                    AND created_at > NOW() - INTERVAL '7 days'
                """)
            finally:
                await close_database_connection(conn3)
            if stuck_orders:
                stuck_ids = [str(o['id']) for o in stuck_orders]
                logger.warning(f"ACME RENEWAL WARNING: {len(stuck_orders)} order(s) stuck > 24h: IDs=[{', '.join(stuck_ids)}]")

        except Exception as e:
            logger.error(f"Error in ACME renewal check: {e}")
            if conn:
                try:
                    await close_database_connection(conn)
                except Exception:
                    pass
        await asyncio.sleep(3600)


# Background task for agent upgrade timeout cleanup
async def cleanup_stuck_agent_upgrades():
    """Background task to reset agents stuck in 'upgrading' status
    
    Agents stuck in 'upgrading' for more than 5 minutes are automatically reset.
    This prevents UI issues where delete button is hidden due to stuck upgrade status.
    """
    while True:
        try:
            conn = await get_database_connection()
            
            # MIGRATION SAFETY: Check if upgrade_status column exists before using it
            # This prevents errors during initial deployment before migration runs
            upgrade_column_exists = await conn.fetchval("""
                SELECT EXISTS (
                    SELECT 1 FROM information_schema.columns 
                    WHERE table_name = 'agents' AND column_name = 'upgrade_status'
                )
            """)
            
            if not upgrade_column_exists:
                await close_database_connection(conn)
                logger.debug("Agent upgrade columns not yet created - skipping cleanup (migration pending)")
                await asyncio.sleep(120)
                continue
            
            # Reset stuck upgrades older than 5 minutes
            # upgraded_at tracks the last upgrade status change
            threshold_time = datetime.utcnow() - timedelta(minutes=5)
            
            result = await conn.execute("""
                UPDATE agents 
                SET upgrade_status = NULL, 
                    upgrade_target_version = NULL,
                    status = 'online',
                    updated_at = CURRENT_TIMESTAMP
                WHERE upgrade_status = 'upgrading' 
                AND (upgraded_at IS NULL OR upgraded_at < $1)
            """, threshold_time)
            
            await close_database_connection(conn)
            
            # Log if any stuck upgrades were reset
            if result and hasattr(result, 'split') and len(result.split()) > 1:
                count = result.split()[1]
                if int(count) > 0:
                    logger.warning(f"Reset {count} stuck agent upgrade(s) - timeout after 5 minutes")
            
        except Exception as e:
            logger.error(f"Error in agent upgrade cleanup: {e}")
        
        # Wait 120 seconds (2 minutes) before next check
        await asyncio.sleep(120)

# Production middleware stack (order matters!)
app.add_middleware(PerformanceMonitoringMiddleware, slow_request_threshold_ms=1000)
app.add_middleware(RequestLoggingMiddleware, exclude_paths=["/api/health/", "/docs", "/redoc"])

# Activity logging middleware - must be before CORS
app.middleware("http")(log_activity_middleware)

# CORS middleware  
app.add_middleware(
    CORSMiddleware,
    allow_origins=CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Global exception handlers
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return await GlobalExceptionHandler.handle_http_exception(request, exc)

@app.exception_handler(StarletteHTTPException)
async def starlette_http_exception_handler(request: Request, exc: StarletteHTTPException):
    return await GlobalExceptionHandler.handle_http_exception(request, HTTPException(status_code=exc.status_code, detail=str(exc.detail)))

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return await GlobalExceptionHandler.handle_validation_error(request, exc)

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return await GlobalExceptionHandler.handle_generic_exception(request, exc)

# Include routers
app.include_router(health_router)  # Health checks first for monitoring
app.include_router(config_router)  # Configuration management
app.include_router(maintenance_router, prefix="/api", tags=["maintenance"])  # Database cleanup & maintenance
app.include_router(auth_router)
app.include_router(user_router)
app.include_router(frontend_router)
app.include_router(backend_router)
app.include_router(cluster_router)
app.include_router(dashboard_router)
app.include_router(dashboard_stats_router)  # HAProxy stats dashboard
app.include_router(agent_router)
app.include_router(waf_router)
app.include_router(ssl_router)
app.include_router(security_router)
app.include_router(configuration_router)
app.include_router(settings_router)
app.include_router(letsencrypt_router)

@app.on_event("startup")
async def startup_event():
    """Initialize the application on startup"""
    logger.info("HAProxy OpenManager API starting up...")
    
    try:
        # Initialize database connection pool FIRST (before any DB operations)
        logger.info("Initializing database connection pool...")
        await init_database_pool()
        logger.info("✅ Database connection pool initialized successfully")
        
        # Run database migrations
        logger.info("Forcing re-check of database schema...")
        await run_all_migrations()
        logger.info("Database migrations completed successfully")
        
        # Initialize agent scripts in database ONLY if empty (first deployment)
        # User edits are preserved, only Reset to Default overwrites from files
        try:
            import os
            
            conn = await get_database_connection()
            
            # Check if agent_script_templates table has any active scripts
            script_count = await conn.fetchval("""
                SELECT COUNT(*) FROM agent_script_templates WHERE is_active = true
            """)
            
            if script_count == 0:
                logger.info("🆕 INITIAL SETUP: No agent scripts in database, loading from files...")
                
                script_files = {
                    'linux': 'linux_install.sh',
                    'macos': 'macos_install.sh'
                }
                
                for platform_key, filename in script_files.items():
                    script_path = os.path.join(os.path.dirname(__file__), 'utils', 'agent_scripts', filename)
                    
                    if os.path.exists(script_path):
                        with open(script_path, 'r') as f:
                            file_content = f.read()
                        
                        # Create initial version 1.0.0
                        await conn.execute("""
                            INSERT INTO agent_script_templates (platform, version, script_content, is_active)
                            VALUES ($1, '1.0.0', $2, true)
                            ON CONFLICT (platform, version) DO NOTHING
                        """, platform_key, file_content)
                        
                        # Also create version entry
                        await conn.execute("""
                            INSERT INTO agent_versions (platform, version, is_active)
                            VALUES ($1, '1.0.0', true)
                            ON CONFLICT (platform, version) DO NOTHING
                        """, platform_key)
                        
                        logger.info(f"✅ INITIAL SETUP: {platform_key} script loaded (version 1.0.0)")
                
                logger.info("✅ INITIAL SETUP: Agent scripts initialized from files")
            else:
                logger.info(f"ℹ️  Agent scripts already exist in database ({script_count} active), preserving user edits")
            
            await close_database_connection(conn)
            
        except Exception as init_error:
            logger.warning(f"Agent script initialization check failed (non-critical): {init_error}")
        
        # Test Redis connection (graceful if unavailable)
        try:
            redis_client.ping()
            logger.info("Redis connection established")
        except Exception as redis_error:
            logger.warning(f"Redis is unavailable, continuing in degraded mode: {redis_error}")
        
        # Start background agent status monitoring task
        asyncio.create_task(monitor_agent_status())
        logger.info("Agent status monitoring task started")
        
        # Start background agent upgrade cleanup task
        asyncio.create_task(cleanup_stuck_agent_upgrades())
        logger.info("Agent upgrade timeout cleanup task started (5 minute timeout)")
        
        # Start ACME certificate auto-renewal task
        asyncio.create_task(check_letsencrypt_renewals())
        logger.info("ACME certificate auto-renewal task started (hourly checks)")
        
        # Create test activity log entry to verify system is working
        try:
            from utils.activity_log import log_user_activity
            await log_user_activity(
                user_id=1,  # Admin user
                action='system_startup',
                resource_type='system',
                resource_id='main',
                details={'event': 'HAProxy OpenManager system started', 'version': _version_info["version"]},
                ip_address='127.0.0.1',
                user_agent='HAProxy-OpenManager-System'
            )
            logger.info("✅ Test activity log created successfully")
        except Exception as log_error:
            logger.error(f"❌ Failed to create test activity log: {log_error}")
        
        logger.info("HAProxy OpenManager API startup completed successfully")
        
    except Exception as e:
        logger.error(f"Failed to initialize application: {e}")
        raise

@app.on_event("shutdown")
async def shutdown_event():
    """Cleanup on shutdown"""
    logger.info("HAProxy OpenManager API shutting down...")
    
    # Close database connection pool gracefully
    try:
        logger.info("Closing database connection pool...")
        await close_database_pool()
        logger.info("✅ Database connection pool closed successfully")
    except Exception as e:
        logger.error(f"❌ Error closing database connection pool: {e}")

@app.get("/")
async def root():
    """Root endpoint"""
    return {"message": "HAProxy Management UI API", "version": _version_info["version"]}

@app.get("/api/version")
async def get_version():
    """Return application version information"""
    return _version_info

@app.get("/.well-known/acme-challenge/{token}")
async def serve_acme_challenge(token: str):
    """Serve ACME HTTP-01 challenge token. Public endpoint, no auth required."""
    logger.info(f"ACME-CHALLENGE: Incoming request for token={token[:32]}...")
    conn = None
    try:
        conn = await get_database_connection()
        row = await conn.fetchrow(
            "SELECT key_authorization FROM acme_challenges WHERE token = $1 AND (status IN ('pending', 'processing') OR status IS NULL) LIMIT 1",
            token
        )
        if row:
            logger.info(f"ACME-CHALLENGE: Token found, serving key_authorization ({len(row['key_authorization'])} chars)")
            from fastapi.responses import PlainTextResponse
            return PlainTextResponse(row['key_authorization'])
        logger.warning(f"ACME-CHALLENGE: Token NOT found in DB - no matching record with status pending/processing/null")
        raise HTTPException(status_code=404, detail="Challenge not found")
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"ACME-CHALLENGE: Error serving token {token[:32]}...: {e}")
        raise HTTPException(status_code=404, detail="Challenge not found")
    finally:
        if conn:
            await close_database_connection(conn)

@app.get("/api/health")
async def health_check():
    """Health check endpoint"""
    try:
        # Check Redis connection
        try:
            redis_client.ping()
            redis_status = "connected"
        except Exception:
            redis_status = "disconnected"
        
        # Check database connection (basic check)
        conn = await get_database_connection()
        await close_database_connection(conn)
        
        return {
            "status": "healthy",
            "timestamp": datetime.utcnow().isoformat(),
            "services": {
                "redis": redis_status,
                "database": "connected"
            }
        }
    except Exception as e:
        return JSONResponse(
            status_code=503,
            content={
                "status": "unhealthy",
                "error": str(e),
                "timestamp": datetime.utcnow().isoformat()
            }
        )