Fixed formatting, lints and typing

app.py

-from flask import Flask, render_template, jsonify, Response, request, redirect, url_for, flash
+from flask import (
+    Flask,
+    render_template,
+    jsonify,
+    request,
+    redirect,
+    url_for,
+    flash,
+)
 from flask_socketio import SocketIO
-from flask_login import LoginManager, login_user, login_required, logout_user, current_user
+from flask_login import (  # type: ignore
+    LoginManager,
+    login_user,
+    login_required,
+    logout_user,
+    current_user,
+)
 from flask_limiter import Limiter
 from flask_limiter.util import get_remote_address
+from flask.typing import ResponseReturnValue
 from pyghmi.ipmi import command  # type: ignore
 import json
 import time
@@ -11,25 +26,27 @@ import logging
 import socket
 import sys
 import argparse
-from typing import Dict, List, Optional, Tuple, TypedDict, Union
-from auth import LDAPAuth, User
+from typing import Dict, List, Optional, Tuple, TypedDict
+from auth import LDAPAuth, User, LDAPConfig
 import os
 from datetime import timedelta
+from typing import Any
 
 # Configure logging to show all levels
 logging.basicConfig(
     level=logging.DEBUG,
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
-    stream=sys.stdout
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    stream=sys.stdout,
 )
 logger = logging.getLogger(__name__)
 
 # Silence Werkzeug logs
-logging.getLogger('werkzeug').setLevel(logging.WARNING)
+logging.getLogger("werkzeug").setLevel(logging.WARNING)
 
 # Set socket timeout
 socket.setdefaulttimeout(5)  # 5 seconds timeout
 
+
 class ServerConfig(TypedDict):
     name: str
     ipmi_ip: str
@@ -37,6 +54,7 @@ class ServerConfig(TypedDict):
     ipmi_pass: str
     locked: bool
 
+
 class ServerStatus(TypedDict):
     name: str
     status: str
@@ -45,54 +63,55 @@ class ServerStatus(TypedDict):
     power_consumption: Optional[float]
     locked: bool
 
+
 class GroupConfig(TypedDict):
     name: str
     servers: List[ServerConfig]
 
-class LDAPConfig(TypedDict):
-    server: str
-    base_dn: str
-    user_dn: str
-    group_dn: str
-    admin_group: str
 
 class Config(TypedDict):
     ldap: LDAPConfig
     groups: List[GroupConfig]
 
+
 class ServerData(TypedDict):
     config: ServerConfig
     connection: Optional[command.Command]
     last_error: Optional[str]
 
+
 class ServerGroup(TypedDict):
     name: str
     servers: List[ServerConfig]
 
+
 class GroupStatus(TypedDict):
     name: str
     servers: List[ServerStatus]
 
+
 app = Flask(__name__)
 app.secret_key = os.urandom(24)
-app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(hours=1)  # Session timeout after 1 hour
+app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(
+    hours=1
+)  # Session timeout after 1 hour
 # app.config['SESSION_COOKIE_SECURE'] = True  # Only send cookies over HTTPS
-app.config['SESSION_COOKIE_HTTPONLY'] = True  # Prevent JavaScript access to cookies
-app.config['SESSION_COOKIE_SAMESITE'] = 'Strict'
+app.config["SESSION_COOKIE_HTTPONLY"] = True  # Prevent JavaScript access to cookies
+app.config["SESSION_COOKIE_SAMESITE"] = "Strict"
 
 socketio = SocketIO(app, cors_allowed_origins="*")
 
 # Initialize LoginManager
 login_manager = LoginManager()
 login_manager.init_app(app)
-login_manager.login_view = 'login'
+login_manager.login_view = "login"
 
 # Load configuration
-with open('config.json', 'r') as f:
+with open("config.json", "r") as f:
     config: Config = json.load(f)
 
 # Initialize LDAP authentication
-ldap_auth = LDAPAuth(config['ldap'])
+ldap_auth = LDAPAuth(config["ldap"])
 
 # Initialize rate limiter
 limiter = Limiter(
@@ -100,13 +119,15 @@ limiter = Limiter(
     key_func=get_remote_address,
     default_limits=["200 per day", "50 per hour"],
     storage_uri="memory://",
-    strategy="fixed-window" 
+    strategy="fixed-window",
 )
 
+
 @login_manager.user_loader
 def load_user(user_id: str) -> Optional[User]:
     return User(user_id)
 
+
 class ServerManager:
     def __init__(self, config: Config) -> None:
         self.servers: Dict[str, ServerData] = {}
@@ -118,17 +139,19 @@ class ServerManager:
 
     def load_config(self, config: Config) -> None:
         try:
-            self.groups = config['groups']
+            self.groups = config["groups"]
             ipmi_ips = set()
             for group in self.groups:
-                for server in group['servers']:
-                    if server['ipmi_ip'] in ipmi_ips:
-                        raise ValueError(f"Duplicate IPMI IP address found in config: {server['ipmi_ip']}")
-                    ipmi_ips.add(server['ipmi_ip'])
+                for server in group["servers"]:
+                    if server["ipmi_ip"] in ipmi_ips:
+                        raise ValueError(
+                            f"Duplicate IPMI IP address found in config: {server['ipmi_ip']}"
+                        )
+                    ipmi_ips.add(server["ipmi_ip"])
                     self.servers[f"{group['name']}/{server['name']}"] = {
-                        'config': server,
-                        'connection': None,
-                        'last_error': None
+                        "config": server,
+                        "connection": None,
+                        "last_error": None,
                     }
         except Exception as e:
             logger.error(f"Failed to load server configuration: {str(e)}")
@@ -140,27 +163,36 @@ class ServerManager:
         for server_name, server_data in self.servers.items():
             try:
                 logger.debug(f"Initializing connection to {server_name}")
-                server_data['connection'] = command.Command(
-                    bmc=server_data['config']['ipmi_ip'],
-                    userid=server_data['config']['ipmi_user'],
-                    password=server_data['config']['ipmi_pass'],
-                    keepalive=True
+                server_data["connection"] = command.Command(
+                    bmc=server_data["config"]["ipmi_ip"],
+                    userid=server_data["config"]["ipmi_user"],
+                    password=server_data["config"]["ipmi_pass"],
+                    keepalive=True,
                 )
                 logger.debug(f"Successfully initialized connection to {server_name}")
             except Exception as e:
-                logger.error(f"Failed to initialize connection to {server_name}: {str(e)}")
-                server_data['last_error'] = str(e)
-
-    def _create_status_response(self, group_name: str, server_name: str, status: str, error: Optional[str] = None, power_consumption: Optional[float] = None) -> ServerStatus:
+                logger.error(
+                    f"Failed to initialize connection to {server_name}: {str(e)}"
+                )
+                server_data["last_error"] = str(e)
+
+    def _create_status_response(
+        self,
+        group_name: str,
+        server_name: str,
+        status: str,
+        error: Optional[str] = None,
+        power_consumption: Optional[float] = None,
+    ) -> ServerStatus:
         """Create a standardized status response for a server."""
         server: ServerData = self.servers[f"{group_name}/{server_name}"]
         return {
-            'name': server_name,
-            'status': status,
-            'error': error,
-            'ipmi_ip': server['config']['ipmi_ip'],
-            'power_consumption': power_consumption,
-            'locked': server['config'].get('locked', False)
+            "name": server_name,
+            "status": status,
+            "error": error,
+            "ipmi_ip": server["config"]["ipmi_ip"],
+            "power_consumption": power_consumption,
+            "locked": server["config"].get("locked", False),
         }
 
     def _update_server_status(self, group_name: str, server_name: str) -> None:
@@ -169,50 +201,53 @@ class ServerManager:
             return
 
         server: ServerData = self.servers[cache_key]
-        if not server['connection']:
+        if not server["connection"]:
             self.status_cache[cache_key] = self._create_status_response(
                 group_name,
                 server_name,
-                'unknown',
-                server['last_error'] or 'No connection established'
+                "unknown",
+                server["last_error"] or "No connection established",
             )
             return
 
         try:
-            status = server['connection'].get_power()
-            server['last_error'] = None
+            status = server["connection"].get_power()
+            server["last_error"] = None
 
             # Get power consumption
             power_consumption: Optional[float] = None
             try:
-                power_consumption = server['connection'].get_system_power_watts()
-                logger.debug(f"Power consumption for {server_name}: {power_consumption}W")
+                power_consumption = server["connection"].get_system_power_watts()
+                logger.debug(
+                    f"Power consumption for {server_name}: {power_consumption}W"
+                )
             except Exception as e:
-                logger.debug(f"Could not get power consumption for {server_name}: {str(e)}")
+                logger.debug(
+                    f"Could not get power consumption for {server_name}: {str(e)}"
+                )
 
             self.status_cache[cache_key] = self._create_status_response(
                 group_name,
                 server_name,
-                'on' if status['powerstate'] == 'on' else 'off',
-                power_consumption=power_consumption
+                "on" if status["powerstate"] == "on" else "off",
+                power_consumption=power_consumption,
             )
         except Exception as e:
             logger.error(f"Failed to get power status for {server_name}: {str(e)}")
-            server['last_error'] = str(e)
+            server["last_error"] = str(e)
             self.status_cache[cache_key] = self._create_status_response(
-                group_name,
-                server_name,
-                'unknown',
-                str(e)
+                group_name, server_name, "unknown", str(e)
             )
 
     def _status_update_loop(self) -> None:
         while True:
             try:
                 for group in self.groups:
-                    for server in group['servers']:
-                        self._update_server_status(group['name'], server['name'])
-                socketio.emit('server_status_update', {'groups': self.get_all_statuses()})
+                    for server in group["servers"]:
+                        self._update_server_status(group["name"], server["name"])
+                socketio.emit(
+                    "server_status_update", {"groups": self.get_all_statuses()}
+                )
             except Exception as e:
                 logger.error(f"Error in status update thread: {str(e)}")
             time.sleep(5)
@@ -221,32 +256,40 @@ class ServerManager:
         thread = threading.Thread(target=self._status_update_loop, daemon=True)
         thread.start()
 
-    def toggle_power(self, group_name: str, server_name: str) -> Tuple[Optional[bool], Optional[str]]:
+    def toggle_power(
+        self, group_name: str, server_name: str
+    ) -> Tuple[Optional[bool], Optional[str]]:
         cache_key = f"{group_name}/{server_name}"
         if cache_key not in self.servers:
-            logger.error(f"Power toggle failed: Server {server_name} not found in group {group_name}")
+            logger.error(
+                f"Power toggle failed: Server {server_name} not found in group {group_name}"
+            )
             return None, "Server not found"
 
         server = self.servers[cache_key]
-        if server['config'].get('locked', False):
+        if server["config"].get("locked", False):
             logger.info(f"Power toggle blocked: Server {server_name} is locked")
             return None, "Server is locked and cannot be controlled via this tool"
 
-        if not server['connection']:
-            logger.error(f"Power toggle failed: No connection established for server {server_name}")
+        if not server["connection"]:
+            logger.error(
+                f"Power toggle failed: No connection established for server {server_name}"
+            )
             return None, "No connection established"
 
         try:
-            current_status = server['connection'].get_power()
-            if current_status['powerstate'] == 'on':
+            current_status = server["connection"].get_power()
+            if current_status["powerstate"] == "on":
                 logger.info(f"Initiating power off for server {server_name}")
-                server['connection'].set_power('off')
+                server["connection"].set_power("off")
             else:
                 logger.info(f"Initiating power on for server {server_name}")
-                server['connection'].set_power('on')
+                server["connection"].set_power("on")
             return True, None
         except Exception as e:
-            logger.error(f"Failed to initiate power toggle for server {server_name}: {str(e)}")
+            logger.error(
+                f"Failed to initiate power toggle for server {server_name}: {str(e)}"
+            )
             return None, str(e)
 
     def get_all_statuses(self) -> List[GroupStatus]:
@@ -255,78 +298,88 @@ class ServerManager:
 
         for group in self.groups:
             group_servers: List[ServerStatus] = []
-            for server in group['servers']:
+            for server in group["servers"]:
                 cache_key = f"{group['name']}/{server['name']}"
                 if cache_key in self.status_cache:
                     group_servers.append(self.status_cache[cache_key])
 
-            group_statuses.append({
-                'name': group['name'],
-                'servers': group_servers
-            })
+            group_statuses.append({"name": group["name"], "servers": group_servers})
 
         return group_statuses
 
+
 config_path = "config.json"
-if __name__ == '__main__':
-    parser = argparse.ArgumentParser(description='IPMI Server Management')
-    parser.add_argument('-c', '--config', default='config.json', help='Path to the configuration file')
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="IPMI Server Management")
+    parser.add_argument(
+        "-c", "--config", default="config.json", help="Path to the configuration file"
+    )
     args = parser.parse_args()
 
     config_path = args.config
 
-with open(config_path, 'r') as f:
+with open(config_path, "r") as f:
     config = json.load(f)
 
 server_manager = ServerManager(config)
 
-@app.route('/login', methods=['GET', 'POST'])
+
+@app.route("/login", methods=["GET", "POST"])
 @limiter.limit("5 per minute")  # Rate limit login attempts
-def login():
-    if request.method == 'POST':
-        username = request.form.get('username')
-        password = request.form.get('password')
+def login() -> ResponseReturnValue:
+    if request.method == "POST":
+        username = request.form.get("username")
+        password = request.form.get("password")
 
         if not username or not password:
-            flash('Please enter both username and password')
-            return redirect(url_for('login'))
+            flash("Please enter both username and password")
+            return redirect(url_for("login"))
 
         user = ldap_auth.authenticate(username, password)
         if user:
             login_user(user)
-            return redirect(url_for('index'))
+            return redirect(url_for("index"))
         else:
-            flash('Invalid username or password')
+            flash("Invalid username or password")
+
+    return render_template("login.html")
 
-    return render_template('login.html')
 
-@app.route('/logout')
+@app.route("/logout")
 @login_required
-def logout():
+def logout() -> ResponseReturnValue:
     logout_user()
-    return redirect(url_for('login'))
+    return redirect(url_for("login"))
 
-@app.route('/')
+
+@app.route("/")
 def index() -> str:
-    return render_template('index.html')
+    return render_template("index.html")
+
 
-@app.route('/api/servers', methods=['GET'])
+@app.route("/api/servers", methods=["GET"])
 @limiter.limit("90 per minute")  # Rate limit server status checks
-def get_servers() -> Response:
-    return jsonify({'groups': server_manager.get_all_statuses()})
+def get_servers() -> Any:
+    return jsonify({"groups": server_manager.get_all_statuses()})
 
-@app.route('/api/servers/<group_name>/<server_name>/power', methods=['POST'])
+
+@app.route("/api/servers/<group_name>/<server_name>/power", methods=["POST"])
 @login_required
-@limiter.limit("4 per minute", key_func=lambda: f"{request.remote_addr}:{request.view_args['group_name']}:{request.view_args['server_name']}")
-def power_action(group_name: str, server_name: str) -> Union[Response, Tuple[Response, int]]:
+@limiter.limit(
+    "4 per minute",
+    key_func=lambda: f"{request.remote_addr}:{request.view_args['group_name']}:{request.view_args['server_name']}",
+)
+def power_action(group_name: str, server_name: str) -> Any:
     success, error = server_manager.toggle_power(group_name, server_name)
     if not success:
-        return jsonify({'error': error}), 500
-    return jsonify({'status': 'success'})
+        return jsonify({"error": error}), 500
+    return jsonify({"status": "success"})
+
+
+@app.route("/api/auth/status", methods=["GET"])
+def auth_status() -> Any:
+    return jsonify({"authenticated": current_user.is_authenticated})
 
-@app.route('/api/auth/status', methods=['GET'])
-def auth_status() -> Response:
-    return jsonify({'authenticated': current_user.is_authenticated})
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     socketio.run(app, debug=True)

auth.py

-import ldap
-from flask_login import UserMixin
-from typing import Optional, Dict, Any
+import ldap  # type: ignore
+from flask_login import UserMixin  # type: ignore
+from typing import Optional, TypedDict
 import logging
 
 logger = logging.getLogger(__name__)
 
+
+class LDAPConfig(TypedDict):
+    server: str
+    base_dn: str
+    user_dn: str
+    user_password: str
+    admin_group: str
+
 class User(UserMixin):
     def __init__(self, username: str):
         self.id = username
 
+
 class LDAPAuth:
-    def __init__(self, config: Dict[str, Any]):
-        self.server = config['server']
-        self.base_dn = config['base_dn']
-        self.user_dn = config['user_dn']
-        self.admin_group = config['admin_group']
-        self.user_password = config['user_password']
+    def __init__(self, config: 'LDAPConfig'):
+        self.server = config["server"]
+        self.base_dn = config["base_dn"]
+        self.user_dn = config["user_dn"]
+        self.admin_group = config["admin_group"]
+        self.user_password = config["user_password"]
         self._service_conn = self._establish_service_connection()
 
     def _establish_service_connection(self) -> ldap.ldapobject.LDAPObject:
@@ -35,10 +44,14 @@ class LDAPAuth:
         try:
             # Search for the user using service connection
             search_filter = f"(uid={username})"
-            result = self._service_conn.search_s(self.base_dn, ldap.SCOPE_SUBTREE, search_filter, ['dn', 'memberOf'])
+            result = self._service_conn.search_s(
+                self.base_dn, ldap.SCOPE_SUBTREE, search_filter, ["dn", "memberOf"]
+            )
 
             if not result:
-                logger.warning(f"Unsuccessful login attempt from user {username}: not found in LDAP")
+                logger.warning(
+                    f"Unsuccessful login attempt from user {username}: not found in LDAP"
+                )
                 return None
 
             # Get user's DN and try to bind with user's credentials
@@ -55,20 +68,25 @@ class LDAPAuth:
                 return None
 
             # Check if user is authorized
-            if 'memberOf' in result[0][1]:
-                for group_dn in result[0][1]['memberOf']:
-                    if group_dn.decode('utf-8').split(',')[0].split('=')[1] == self.admin_group:
+            if "memberOf" in result[0][1]:
+                for group_dn in result[0][1]["memberOf"]:
+                    if (
+                        group_dn.decode("utf-8").split(",")[0].split("=")[1]
+                        == self.admin_group
+                    ):
                         logger.info(f"Successful login of user {username}")
                         return User(username)
 
-            logger.warning(f"User {username} is not a member of the authorized group {self.admin_group}")
+            logger.warning(
+                f"User {username} is not a member of the authorized group {self.admin_group}"
+            )
             return None
 
         except Exception as e:
             logger.error(f"LDAP authentication error: {str(e)}")
             return None
 
-    def __del__(self):
+    def __del__(self) -> None:
         """Clean up service connection when object is destroyed."""
         if self._service_conn:
             self._service_conn.unbind_s()