""" 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 Curves API @app.get("/api/curves") async def api_list_curves(username: str = Depends(get_current_user)): """List all fan curves.""" service = get_service(str(CONFIG_FILE)) return { "curves": service.config.get('fan_curves', {}), "active": service.config.get('active_curve', 'Balanced') } @app.post("/api/curves") async def api_create_curve(data: dict, username: str = Depends(get_current_user)): """Create or update a fan curve.""" service = get_service(str(CONFIG_FILE)) name = data.get('name') if not name: return {"success": False, "error": "Name required"} curves = service.config.get('fan_curves', {}) curves[name] = { "points": data.get('points', []), "sensor_source": data.get('sensor_source', 'cpu'), "applies_to": data.get('applies_to', 'all') } service.update_config(fan_curves=curves) return {"success": True} @app.post("/api/curves/active") async def api_set_active_curve(data: dict, username: str = Depends(get_current_user)): """Set active fan curve.""" service = get_service(str(CONFIG_FILE)) name = data.get('name') curves = service.config.get('fan_curves', {}) if name not in curves: return {"success": False, "error": "Curve not found"} service.update_config(active_curve=name) return {"success": True} @app.delete("/api/curves/{name}") async def api_delete_curve(name: str, username: str = Depends(get_current_user)): """Delete a fan curve.""" service = get_service(str(CONFIG_FILE)) curves = service.config.get('fan_curves', {}) if name in curves: del curves[name] service.update_config(fan_curves=curves) 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)