""" IPMI Controller Web Server Advanced web interface with dark mode, fan groups, multiple curves """ import asyncio import json import logging import hashlib import secrets from contextlib import asynccontextmanager from pathlib import Path from typing import Optional, List, Dict from datetime import datetime, timedelta from fastapi import FastAPI, HTTPException, Depends, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse, Response from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from pydantic import BaseModel, Field import sys sys.path.insert(0, str(Path(__file__).parent)) from fan_controller import get_service, IPMIControllerService logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) security = HTTPBearer(auto_error=False) # Data directories DATA_DIR = Path("/app/data") if Path("/app/data").exists() else Path(__file__).parent / "data" DATA_DIR.mkdir(exist_ok=True) CONFIG_FILE = DATA_DIR / "config.json" USERS_FILE = DATA_DIR / "users.json" # Pydantic models class UserLogin(BaseModel): username: str password: str class ChangePassword(BaseModel): current_password: str new_password: str class SetupRequest(BaseModel): admin_username: str admin_password: str ipmi_host: str ipmi_username: str ipmi_password: str ipmi_port: int = 623 class IPMIConfig(BaseModel): host: str username: str password: Optional[str] = None port: int = 623 class HTTPConfig(BaseModel): enabled: bool = False url: Optional[str] = None timeout: int = 10 class FanSettings(BaseModel): enabled: Optional[bool] = None poll_interval: Optional[int] = Field(None, ge=5, le=300) fan_update_interval: Optional[int] = Field(None, ge=5, le=300) min_speed: Optional[int] = Field(None, ge=0, le=100) max_speed: Optional[int] = Field(None, ge=0, le=100) panic_temp: Optional[float] = Field(None, ge=50, le=100) panic_speed: Optional[int] = Field(None, ge=0, le=100) panic_on_no_data: Optional[bool] = None no_data_timeout: Optional[int] = Field(None, ge=10, le=300) primary_sensor: Optional[str] = None sensor_preference: Optional[str] = None theme: Optional[str] = None class FanCurvePoint(BaseModel): temp: float = Field(..., ge=0, le=100) speed: int = Field(..., ge=0, le=100) class FanCurveCreate(BaseModel): name: str points: List[FanCurvePoint] sensor_source: str = "cpu" applies_to: str = "all" class FanConfig(BaseModel): fan_id: str name: Optional[str] = None group: Optional[str] = None curve: Optional[str] = None class FanGroupCreate(BaseModel): name: str fans: List[str] curve: str = "Default" class ManualSpeedRequest(BaseModel): speed: int = Field(..., ge=0, le=100) fan_id: str = "0xff" class IdentifyRequest(BaseModel): fan_id: str # User Manager class UserManager: def __init__(self): self.users_file = USERS_FILE self._users = {} self._sessions = {} self._load() def _load(self): if self.users_file.exists(): try: with open(self.users_file) as f: data = json.load(f) self._users = data.get('users', {}) except Exception as e: logger.error(f"Failed to load users: {e}") def _save(self): with open(self.users_file, 'w') as f: json.dump({'users': self._users}, f) def _hash(self, password: str) -> str: return hashlib.sha256(password.encode()).hexdigest() def verify(self, username: str, password: str) -> bool: return username in self._users and self._users[username] == self._hash(password) def create(self, username: str, password: str) -> bool: if username in self._users: return False self._users[username] = self._hash(password) self._save() return True def change_password(self, username: str, current: str, new: str) -> bool: if not self.verify(username, current): return False self._users[username] = self._hash(new) self._save() return True def create_token(self, username: str) -> str: token = secrets.token_urlsafe(32) self._sessions[token] = (username, datetime.utcnow() + timedelta(days=7)) return token def verify_token(self, token: str) -> Optional[str]: if token not in self._sessions: return None username, expiry = self._sessions[token] if datetime.utcnow() > expiry: del self._sessions[token] return None return username def is_setup_complete(self) -> bool: return len(self._users) > 0 user_manager = UserManager() async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)) -> str: if not credentials: raise HTTPException(status_code=401, detail="Not authenticated") username = user_manager.verify_token(credentials.credentials) if not username: raise HTTPException(status_code=401, detail="Invalid or expired token") return username # CSS Themes THEMES = { "dark": """ :root { --bg-primary: #0f0f1e; --bg-secondary: #1a1a2e; --bg-card: rgba(255,255,255,0.05); --text-primary: #ffffff; --text-secondary: #a0a0a0; --accent-primary: #2196f3; --accent-success: #4caf50; --accent-warning: #ff9800; --accent-danger: #f44336; --border: rgba(255,255,255,0.1); } """, "light": """ :root { --bg-primary: #f5f5f5; --bg-secondary: #ffffff; --bg-card: #ffffff; --text-primary: #333333; --text-secondary: #666666; --accent-primary: #1976d2; --accent-success: #388e3c; --accent-warning: #f57c00; --accent-danger: #d32f2f; --border: #e0e0e0; } body { background: var(--bg-primary) !important; color: var(--text-primary) !important; } .card { background: var(--bg-card) !important; border-color: var(--border) !important; } input, select, textarea { background: #f0f0f0 !important; color: #333 !important; border-color: #ccc !important; } """ } # HTML Template def get_html(theme="dark"): theme_css = THEMES.get(theme, THEMES["dark"]) return f''' IPMI Controller

🌡️ IPMI Controller

🖥️
IPMI
-
⚙️
Mode
-
🌡️
Max Temp
-
🌬️
Fan Speed
-
📊
Sensors
-

🎛️ Quick Controls

🌡️ Temperatures

Loading...

🌬️ Fans

Loading...

IPMI Configuration

HTTP Sensor (lm-sensors over HTTP)

Fan Control Settings

Fan Configuration

Connect to IPMI to see fans

Fan Groups

No groups defined

Fan Curves

System Logs

''' LOGIN_HTML = ''' Login - IPMI Controller

🌡️ IPMI Controller

''' SETUP_HTML = ''' Setup - IPMI Controller

🌡️ IPMI Controller

Initial Setup

👤 Admin Account

🖥️ IPMI Connection

''' # FastAPI app @asynccontextmanager async def lifespan(app: FastAPI): DATA_DIR.mkdir(parents=True, exist_ok=True) yield app = FastAPI( title="IPMI Controller", description="Advanced fan control for Dell servers", version="3.0.0", lifespan=lifespan ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Routes @app.get("/favicon.ico") async def favicon(): """Transparent 1x1 PNG favicon.""" return Response( content=bytes([ 0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A, 0x00, 0x00, 0x00, 0x0D, 0x49, 0x48, 0x44, 0x52, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x01, 0x08, 0x06, 0x00, 0x00, 0x00, 0x1F, 0x15, 0xC4, 0x89, 0x00, 0x00, 0x00, 0x0D, 0x49, 0x44, 0x41, 0x54, 0x08, 0xD7, 0x63, 0x60, 0x60, 0x60, 0x00, 0x00, 0x00, 0x02, 0x00, 0x01, 0xE2, 0x21, 0xBC, 0x33, 0x00, 0x00, 0x00, 0x00, 0x49, 0x45, 0x4E, 0x44, 0xAE, 0x42, 0x60, 0x82 ]), media_type="image/png" ) @app.get("/") async def root(request: Request): """Main dashboard.""" if not user_manager.is_setup_complete(): return HTMLResponse(content=SETUP_HTML) # Get theme preference from query or default to dark theme = request.query_params.get('theme', 'dark') return HTMLResponse(content=get_html(theme)) @app.get("/login") async def login_page(): """Login page.""" if not user_manager.is_setup_complete(): return HTMLResponse(content=SETUP_HTML) return HTMLResponse(content=LOGIN_HTML) # Auth API @app.post("/api/auth/login") async def api_login(credentials: UserLogin): if not user_manager.verify(credentials.username, credentials.password): return {"success": False, "error": "Invalid credentials"} token = user_manager.create_token(credentials.username) return {"success": True, "token": token} @app.post("/api/auth/change-password") async def api_change_password(data: ChangePassword, username: str = Depends(get_current_user)): if not user_manager.change_password(username, data.current_password, data.new_password): return {"success": False, "error": "Current password is incorrect"} return {"success": True} @app.post("/api/setup") async def api_setup(data: SetupRequest): if user_manager.is_setup_complete(): return {"success": False, "error": "Setup already completed"} if not user_manager.create(data.admin_username, data.admin_password): return {"success": False, "error": "Failed to create user"} service = get_service(str(CONFIG_FILE)) service.update_config( ipmi_host=data.ipmi_host, ipmi_username=data.ipmi_username, ipmi_password=data.ipmi_password, ipmi_port=data.ipmi_port ) token = user_manager.create_token(data.admin_username) return {"success": True, "token": token} # Status API @app.get("/api/status") async def api_status(username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) return service.get_status() @app.post("/api/test") async def api_test(username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) if not service.controller: if not service._init_controller(): return {"success": False, "error": "Failed to connect - check config"} success = service.controller.test_connection() return {"success": success, "error": None if success else "Connection failed"} # Control API @app.post("/api/control/auto") async def api_control_auto(data: dict, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) service.set_auto_mode(data.get('enabled', False)) if data.get('enabled') and not service.running: if not service.start(): return {"success": False, "error": "Failed to start"} return {"success": True} @app.post("/api/control/manual") async def api_control_manual(req: ManualSpeedRequest, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) if not service.controller: if not service._init_controller(): return {"success": False, "error": "Not connected"} if service.set_manual_speed(req.speed, req.fan_id): return {"success": True} return {"success": False, "error": "Failed"} # Config API @app.post("/api/config/ipmi") async def api_config_ipmi(data: IPMIConfig, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) updates = { "ipmi_host": data.host, "ipmi_username": data.username, "ipmi_port": data.port } if data.password: updates["ipmi_password"] = data.password service.update_config(**updates) return {"success": True} @app.post("/api/config/http") async def api_config_http(data: HTTPConfig, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) service.update_config( http_sensor_enabled=data.enabled, http_sensor_url=data.url, http_sensor_timeout=data.timeout ) return {"success": True} @app.post("/api/config/settings") async def api_config_settings(data: FanSettings, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) updates = {k: v for k, v in data.model_dump().items() if v is not None} service.update_config(**updates) return {"success": True} # Fan API @app.post("/api/fans/identify") async def api_identify_fan(req: IdentifyRequest, username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) if service.identify_fan(req.fan_id): return {"success": True} return {"success": False, "error": "Failed"} @app.post("/api/fans/stop-identify") async def api_stop_identify(username: str = Depends(get_current_user)): service = get_service(str(CONFIG_FILE)) service.stop_identify() return {"success": True} # Public API (no auth required - for external integrations) @app.get("/api/public/status") async def api_public_status(): """Public status endpoint for integrations.""" service = get_service(str(CONFIG_FILE)) status = service.get_status() # Return limited public data return { "temperatures": status.get("temperatures", []), "fans": status.get("fans", []), "current_speeds": status.get("current_speeds", {}), "connected": status.get("connected", False), "enabled": status.get("enabled", False) } @app.get("/api/public/temperatures") async def api_public_temps(): """Public temperatures endpoint.""" service = get_service(str(CONFIG_FILE)) return {"temperatures": service.get_status().get("temperatures", [])} @app.get("/api/public/fans") async def api_public_fans(): """Public fans endpoint.""" service = get_service(str(CONFIG_FILE)) return {"fans": service.get_status().get("fans", [])} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)