from __future__ import annotations import os import shutil import time import subprocess from logger import get_logger log = get_logger("actions.nginx") NGINX_BASE_DIR = "/etc/nginx" def _verify_safe_path(config_path: str) -> str: """Resolve the real path and verify it is strictly within NGINX_BASE_DIR to prevent path traversal.""" if not config_path: raise ValueError("config_path is required") resolved_path = os.path.realpath(config_path) # Check against realpath of NGINX_BASE_DIR to ensure cross-platform compatibility nginx_base = os.path.realpath(NGINX_BASE_DIR) if not resolved_path.startswith(nginx_base + os.sep) and resolved_path != nginx_base: raise PermissionError(f"Access denied: {config_path} is outside of Nginx config directory") return resolved_path def handle_nginx_config_read(payload: dict) -> dict: """Read Nginx configuration file content securely.""" try: config_path = payload.get("config_path", "") resolved_path = _verify_safe_path(config_path) if not os.path.isfile(resolved_path): raise FileNotFoundError(f"Configuration file not found: {config_path}") file_size = os.path.getsize(resolved_path) if file_size > 512 * 1024: raise ValueError("Configuration file exceeds maximum size of 512KB") with open(resolved_path, "r", encoding="utf-8", errors="ignore") as f: content = f.read() return { "success": True, "content": content, "path": config_path, "size_bytes": file_size } except Exception as exc: log.error("Nginx config read failed: %s", exc) return { "success": False, "error": str(exc) } def handle_nginx_config_write(payload: dict) -> dict: """Write Nginx configuration file content securely with validation, backup, and rollback.""" resolved_path = None backup_path = None orig_temp_path = None original_existed = False try: config_path = payload.get("config_path", "") content = payload.get("content") if content is None: raise ValueError("content is required") # Path traversal protection resolved_path = _verify_safe_path(config_path) # Check size constraints if len(content.encode("utf-8")) > 512 * 1024: raise ValueError("Configuration content exceeds maximum size of 512KB") # Restrict write locations to NGINX_BASE_DIR subdirs allowed_dirs = [ os.path.realpath(os.path.join(NGINX_BASE_DIR, "sites-available")), os.path.realpath(os.path.join(NGINX_BASE_DIR, "conf.d")) ] parent_dir = os.path.dirname(resolved_path) if not any(parent_dir.startswith(d + os.sep) or parent_dir == d for d in allowed_dirs): raise PermissionError("Directory is not allowed for writing configuration files") # ADIM 1: Backup current file (if it exists) original_existed = os.path.exists(resolved_path) if original_existed: backup_path = f"{resolved_path}.wolfpanel.bak.{int(time.time())}" shutil.copy2(resolved_path, backup_path) # ADIM 2: Write new content to temporary file tmp_path = f"{resolved_path}.tmp" with open(tmp_path, "w", encoding="utf-8") as f: f.write(content) # ADIM 3: Temporarily swap original with tmp file to test orig_temp_path = f"{resolved_path}.orig_temp" if original_existed: os.rename(resolved_path, orig_temp_path) os.rename(tmp_path, resolved_path) # Test configuration with nginx -t -c /etc/nginx/nginx.conf try: res = subprocess.run( ["nginx", "-t", "-c", "/etc/nginx/nginx.conf"], capture_output=True, text=True, shell=False, check=False ) test_success = (res.returncode == 0) nginx_output = f"stdout:\n{res.stdout}\nstderr:\n{res.stderr}" except Exception as e: test_success = False nginx_output = f"Nginx test command execution failed: {str(e)}" # ADIM 4 & 5: Resolve based on test outcome if test_success: # Success: delete the temporary backup (orig_temp) if it exists if original_existed and os.path.exists(orig_temp_path): os.remove(orig_temp_path) return { "success": True, "message": "Nginx configuration written successfully and tested OK", "path": config_path } else: # Failure: delete the new invalid config, restore the old config, and return error if os.path.exists(resolved_path): os.remove(resolved_path) if original_existed and os.path.exists(orig_temp_path): os.rename(orig_temp_path, resolved_path) elif backup_path and os.path.exists(backup_path): shutil.copy2(backup_path, resolved_path) return { "success": False, "error": "nginx_test_failed", "nginx_output": nginx_output } except Exception as exc: log.error("Nginx config write failed: %s", exc) # Rollback in case of general failure before/during processing try: if resolved_path: tmp_path = f"{resolved_path}.tmp" if os.path.exists(tmp_path): os.remove(tmp_path) orig_temp_path = f"{resolved_path}.orig_temp" if original_existed and os.path.exists(orig_temp_path): if os.path.exists(resolved_path): os.remove(resolved_path) os.rename(orig_temp_path, resolved_path) except Exception as rollback_exc: log.error("Nginx rollback failed during general exception handler: %s", rollback_exc) return { "success": False, "error": str(exc) } def handle_nginx_service_reload(payload: dict) -> dict: """Reload Nginx service using systemctl reload nginx.""" try: res = subprocess.run( ["systemctl", "reload", "nginx"], capture_output=True, text=True, shell=False, check=False ) if res.returncode == 0: return { "success": True, "reloaded": True, "output": res.stdout or res.stderr or "Nginx service reloaded successfully." } else: status_res = subprocess.run( ["systemctl", "status", "nginx"], capture_output=True, text=True, shell=False, check=False ) combined_output = ( f"Reload failed (code {res.returncode}):\n{res.stderr}\n" f"Service Status:\n{status_res.stdout}\n{status_res.stderr}" ) return { "success": False, "reloaded": False, "error": "reload_failed", "output": combined_output } except Exception as exc: log.error("Nginx reload failed: %s", exc) return { "success": False, "reloaded": False, "error": str(exc), "output": f"Exception occurred while reloading Nginx: {str(exc)}" }