diff --git a/core/__init__.py b/core/__init__.py new file mode 100644 index 0000000..ddf1a64 --- /dev/null +++ b/core/__init__.py @@ -0,0 +1,2 @@ +# -*- coding: utf-8 -*- +"""HestiaCP Django-Gunicorn Deployment Engine — Core Package.""" diff --git a/core/detector.py b/core/detector.py new file mode 100644 index 0000000..f4ccdbf --- /dev/null +++ b/core/detector.py @@ -0,0 +1,207 @@ +# -*- coding: utf-8 -*- +""" +Smart Django project structure detector. + +**Single Responsibility:** Inspect a ``public_html`` directory and produce a +fully populated ``DjangoProjectInfo`` — identifying the WSGI module, virtual +environment, static-file conventions, and database engine. + +Supports three layout styles: + * **Modular** — ``config/wsgi.py`` (clean architecture like ``portfolio``) + * **Traditional** — ``/wsgi.py`` (``django-admin startproject``) + * **Flat** — ``wsgi.py`` at the project root (rare but valid) +""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import Optional + +from core.interfaces import IDjangoDetector +from core.models import ( + DEFAULT_STATIC_DIR, + DEFAULT_VENV_CANDIDATES, + DatabaseEngine, + DjangoProjectInfo, + ProjectLayout, +) + + +class DjangoProjectDetector(IDjangoDetector): + """Concrete detector that inspects the filesystem for Django conventions.""" + + # Regex patterns for extracting settings values + _STATIC_ROOT_RE = re.compile( + r"""STATIC_ROOT\s*=\s*(?:""" + r"""BASE_DIR\s*/\s*['"]([^'"]+)['"]""" # BASE_DIR / 'staticfiles' + r"""|os\.path\.join\s*\(\s*BASE_DIR\s*,\s*['"]([^'"]+)['"]""" # os.path.join(...) + r"""|['"]([^'"]+)['"]""" # plain string + r""")""", + re.VERBOSE, + ) + _DB_ENGINE_RE = re.compile(r"""['"]ENGINE['"]\s*:\s*['"]([^'"]+)['"]""") + _WSGI_APP_RE = re.compile(r"""WSGI_APPLICATION\s*=\s*['"]([^'"]+)['"]""") + + def detect(self, web_dir: Path, project_hint: Optional[str] = None) -> DjangoProjectInfo: + """Analyse *web_dir* and return a populated ``DjangoProjectInfo``.""" + info = DjangoProjectInfo() + + if not web_dir.is_dir(): + info.errors.append(f"Web directory does not exist: {web_dir}") + return info + + # --- 1. Detect project layout & WSGI module --- + self._detect_layout(web_dir, info, project_hint) + + # --- 2. Locate virtualenv --- + self._detect_venv(web_dir, info) + + # --- 3. Inspect settings.py --- + self._inspect_settings(web_dir, info) + + # --- 4. Check for .env and requirements.txt --- + info.has_dot_env = (web_dir / ".env").is_file() + info.has_requirements = (web_dir / "requirements.txt").is_file() + + return info + + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ + + def _detect_layout( + self, + web_dir: Path, + info: DjangoProjectInfo, + project_hint: Optional[str], + ) -> None: + """Determine project layout and set ``wsgi_module``.""" + + # Priority 1: Modular layout (config/wsgi.py) + if (web_dir / "config" / "wsgi.py").is_file(): + info.layout = ProjectLayout.MODULAR + info.wsgi_module = "config.wsgi:application" + info.settings_module = "config.settings" + return + + # Priority 2: Traditional layout (/wsgi.py) + if project_hint: + candidate = web_dir / project_hint / "wsgi.py" + if candidate.is_file(): + info.layout = ProjectLayout.TRADITIONAL + info.wsgi_module = f"{project_hint}.wsgi:application" + info.settings_module = f"{project_hint}.settings" + return + + # Auto-scan for traditional layout directories + for child in sorted(web_dir.iterdir()): + if child.is_dir() and (child / "wsgi.py").is_file(): + name = child.name + info.layout = ProjectLayout.TRADITIONAL + info.wsgi_module = f"{name}.wsgi:application" + info.settings_module = f"{name}.settings" + return + + # Priority 3: Flat layout (wsgi.py at root) + if (web_dir / "wsgi.py").is_file(): + info.layout = ProjectLayout.FLAT + info.wsgi_module = "wsgi:application" + # Try to read DJANGO_SETTINGS_MODULE from wsgi.py + self._extract_settings_from_wsgi(web_dir / "wsgi.py", info) + return + + info.errors.append( + "Could not locate wsgi.py — checked config/, /, and root." + ) + + def _detect_venv(self, web_dir: Path, info: DjangoProjectInfo) -> None: + """Find the first existing virtualenv candidate directory.""" + search_dirs = [web_dir, web_dir.parent] # public_html, then domain root + + for base in search_dirs: + for candidate_name in DEFAULT_VENV_CANDIDATES: + candidate = base / candidate_name + if (candidate / "bin" / "python").is_file() or \ + (candidate / "bin" / "python3").is_file(): + info.venv_path = candidate + return + + info.errors.append( + f"No virtualenv found. Searched for {DEFAULT_VENV_CANDIDATES} " + f"in {web_dir} and {web_dir.parent}." + ) + + def _inspect_settings(self, web_dir: Path, info: DjangoProjectInfo) -> None: + """Parse ``settings.py`` for STATIC_ROOT and DATABASE ENGINE.""" + settings_path = self._find_settings_file(web_dir, info) + if settings_path is None: + return + + try: + content = settings_path.read_text(encoding="utf-8", errors="replace") + except OSError as exc: + info.errors.append(f"Cannot read settings: {exc}") + return + + # STATIC_ROOT + match = self._STATIC_ROOT_RE.search(content) + if match: + # Pick the first non-None group + info.static_root_name = next( + (g for g in match.groups() if g), DEFAULT_STATIC_DIR + ) + + # Database engine + db_match = self._DB_ENGINE_RE.search(content) + if db_match: + engine_str = db_match.group(1).lower() + if "sqlite" in engine_str: + info.database_engine = DatabaseEngine.SQLITE + elif "postgresql" in engine_str or "postgis" in engine_str: + info.database_engine = DatabaseEngine.POSTGRESQL + elif "mysql" in engine_str: + info.database_engine = DatabaseEngine.MYSQL + else: + info.database_engine = DatabaseEngine.OTHER + + def _find_settings_file( + self, web_dir: Path, info: DjangoProjectInfo + ) -> Optional[Path]: + """Locate settings.py based on the detected layout.""" + if info.settings_module: + relative = info.settings_module.replace(".", "/") + ".py" + candidate = web_dir / relative + if candidate.is_file(): + return candidate + + # Fallback: search common locations + for candidate in [ + web_dir / "config" / "settings.py", + web_dir / "settings.py", + ]: + if candidate.is_file(): + return candidate + + # Search any subdirectory + for child in sorted(web_dir.iterdir()): + if child.is_dir(): + settings = child / "settings.py" + if settings.is_file(): + return settings + + return None + + @staticmethod + def _extract_settings_from_wsgi(wsgi_path: Path, info: DjangoProjectInfo) -> None: + """Try to read DJANGO_SETTINGS_MODULE from a flat wsgi.py.""" + try: + content = wsgi_path.read_text(encoding="utf-8", errors="replace") + match = re.search( + r"""os\.environ.*?['"]DJANGO_SETTINGS_MODULE['"]\s*,\s*['"]([^'"]+)['"]""", + content, + ) + if match: + info.settings_module = match.group(1) + except OSError: + pass diff --git a/core/doctor.py b/core/doctor.py new file mode 100644 index 0000000..4702162 --- /dev/null +++ b/core/doctor.py @@ -0,0 +1,287 @@ +# -*- coding: utf-8 -*- +""" +Post-deployment health checker (doctor). + +**Single Responsibility:** Run a suite of diagnostic checks against a live +deployment and return a structured ``DiagnosticReport`` with colour-coded +terminal output. +""" + +from __future__ import annotations + +try: + import grp +except ImportError: + grp = None # type: ignore + +try: + import pwd +except ImportError: + pwd = None # type: ignore + +import os +import stat +from pathlib import Path + +from core.interfaces import IDeploymentDoctor, ISystemExecutor +from core.models import ( + DiagnosticCheck, + DiagnosticReport, + DiagnosticStatus, + DomainConfig, +) + + +class DeploymentDoctor(IDeploymentDoctor): + """Runs comprehensive health checks on a deployed Django/Gunicorn site.""" + + def __init__(self, executor: ISystemExecutor) -> None: + self._executor = executor + + def diagnose(self, domain_config: DomainConfig) -> DiagnosticReport: + """Run all health checks and return a ``DiagnosticReport``.""" + dc = domain_config + report = DiagnosticReport(domain_config=dc) + + self._check_service_status(dc, report) + self._check_socket(dc, report) + self._check_database_permissions(dc, report) + self._check_static_files(dc, report) + self._check_nginx_template(dc, report) + self._check_socket_response(dc, report) + + return report + + # ------------------------------------------------------------------ + # Individual checks + # ------------------------------------------------------------------ + + def _check_service_status(self, dc: DomainConfig, report: DiagnosticReport) -> None: + """Check if the Gunicorn systemd service is active.""" + rc, stdout, _ = self._executor.run( + f"systemctl is-active {dc.service_name}", check=False + ) + if rc == 0 and "active" in stdout: + report.checks.append(DiagnosticCheck( + name="Gunicorn Service", + status=DiagnosticStatus.OK, + message=f"{dc.service_name} is active and running.", + )) + else: + report.checks.append(DiagnosticCheck( + name="Gunicorn Service", + status=DiagnosticStatus.CRITICAL, + message=f"{dc.service_name} is NOT running.", + suggestion=f"sudo systemctl start {dc.service_name}\n" + f"journalctl -u {dc.service_name} -n 30", + )) + + def _check_socket(self, dc: DomainConfig, report: DiagnosticReport) -> None: + """Verify unix socket exists with correct ownership and permissions.""" + sock = dc.socket_path + + if not self._executor.file_exists(sock): + report.checks.append(DiagnosticCheck( + name="Unix Socket", + status=DiagnosticStatus.CRITICAL, + message=f"Socket not found: {sock}", + suggestion=f"sudo systemctl restart {dc.service_name}", + )) + return + + try: + st = os.stat(str(sock)) + owner = pwd.getpwuid(st.st_uid).pw_name + group = grp.getgrgid(st.st_gid).gr_name + mode = stat.S_IMODE(st.st_mode) + + issues = [] + if owner != dc.user: + issues.append(f"Owner is '{owner}', expected '{dc.user}'") + if group != "www-data": + issues.append(f"Group is '{group}', expected 'www-data'") + if mode & 0o060 != 0o060: + issues.append(f"Mode is {oct(mode)}, group needs rw (0o660+)") + + if issues: + report.checks.append(DiagnosticCheck( + name="Unix Socket", + status=DiagnosticStatus.WARNING, + message=f"Socket exists but has permission issues: {'; '.join(issues)}", + suggestion=f"sudo chown {dc.user}:www-data {sock} && sudo chmod 660 {sock}", + )) + else: + report.checks.append(DiagnosticCheck( + name="Unix Socket", + status=DiagnosticStatus.OK, + message=f"Socket OK — {owner}:{group} {oct(mode)}", + )) + except (OSError, KeyError) as exc: + report.checks.append(DiagnosticCheck( + name="Unix Socket", + status=DiagnosticStatus.WARNING, + message=f"Could not inspect socket: {exc}", + )) + + def _check_database_permissions(self, dc: DomainConfig, report: DiagnosticReport) -> None: + """Check db.sqlite3 is writable by www-data group.""" + db_path = dc.web_dir / "db.sqlite3" + + if not self._executor.file_exists(db_path): + report.checks.append(DiagnosticCheck( + name="Database (SQLite)", + status=DiagnosticStatus.OK, + message="No db.sqlite3 found (possibly using PostgreSQL).", + )) + return + + try: + st = os.stat(str(db_path)) + mode = stat.S_IMODE(st.st_mode) + group = grp.getgrgid(st.st_gid).gr_name if grp else "www-data" + + if group != "www-data" or mode & 0o060 != 0o060: + report.checks.append(DiagnosticCheck( + name="Database (SQLite)", + status=DiagnosticStatus.CRITICAL, + message=f"db.sqlite3 permissions incorrect (group={group}, mode={oct(mode)}).", + suggestion=f"sudo chown {dc.user}:www-data {db_path} && sudo chmod 664 {db_path}", + )) + else: + report.checks.append(DiagnosticCheck( + name="Database (SQLite)", + status=DiagnosticStatus.OK, + message="db.sqlite3 permissions correct.", + )) + except (OSError, KeyError): + report.checks.append(DiagnosticCheck( + name="Database (SQLite)", + status=DiagnosticStatus.WARNING, + message="Could not inspect db.sqlite3 permissions.", + )) + + def _check_static_files(self, dc: DomainConfig, report: DiagnosticReport) -> None: + """Check if collectstatic has been run.""" + candidates = ["staticfiles", "static"] + for name in candidates: + static_dir = dc.web_dir / name + if self._executor.file_exists(static_dir) and any(static_dir.iterdir()): + report.checks.append(DiagnosticCheck( + name="Static Files", + status=DiagnosticStatus.OK, + message=f"Static files found in {static_dir.name}/", + )) + return + + report.checks.append(DiagnosticCheck( + name="Static Files", + status=DiagnosticStatus.WARNING, + message="No static files found — Django Admin will have no styling.", + suggestion="cd {web_dir} && source venv/bin/activate && " + "python manage.py collectstatic --noinput".format(web_dir=dc.web_dir), + )) + + def _check_nginx_template(self, dc: DomainConfig, report: DiagnosticReport) -> None: + """Check if NGINX config for this domain points to the correct socket.""" + nginx_conf = Path(f"/home/{dc.user}/conf/web/{dc.domain}/nginx.conf") + + if not self._executor.file_exists(nginx_conf): + report.checks.append(DiagnosticCheck( + name="NGINX Configuration", + status=DiagnosticStatus.WARNING, + message="NGINX domain config not found — template may not be applied.", + suggestion=f"sudo v-rebuild-web-domain {dc.user} {dc.domain} yes", + )) + return + + try: + content = nginx_conf.read_text(encoding="utf-8", errors="replace") + expected_socket = str(dc.socket_path) + if expected_socket in content: + report.checks.append(DiagnosticCheck( + name="NGINX Configuration", + status=DiagnosticStatus.OK, + message="NGINX config points to the correct socket path.", + )) + else: + report.checks.append(DiagnosticCheck( + name="NGINX Configuration", + status=DiagnosticStatus.CRITICAL, + message="NGINX config does NOT point to the expected socket!", + suggestion=f"sudo v-rebuild-web-domain {dc.user} {dc.domain} yes && " + "sudo systemctl restart nginx", + )) + except OSError: + report.checks.append(DiagnosticCheck( + name="NGINX Configuration", + status=DiagnosticStatus.WARNING, + message="Could not read NGINX configuration.", + )) + + def _check_socket_response(self, dc: DomainConfig, report: DiagnosticReport) -> None: + """Try to get an HTTP response through the unix socket.""" + sock = dc.socket_path + if not self._executor.file_exists(sock): + return # Already reported in _check_socket + + rc, stdout, _ = self._executor.run( + f"curl -s -o /dev/null -w '%{{http_code}}' " + f"--unix-socket {sock} http://localhost/ --max-time 5", + check=False, + ) + if rc == 0 and stdout.strip("'\"") in ("200", "301", "302", "403"): + report.checks.append(DiagnosticCheck( + name="Socket Response", + status=DiagnosticStatus.OK, + message=f"Socket responds with HTTP {stdout.strip(chr(39))}.", + )) + else: + report.checks.append(DiagnosticCheck( + name="Socket Response", + status=DiagnosticStatus.WARNING, + message=f"Socket did not respond as expected (rc={rc}, response={stdout}).", + suggestion=f"journalctl -u {dc.service_name} -n 20", + )) + + +# --------------------------------------------------------------------------- +# Pretty-printer for terminal output +# --------------------------------------------------------------------------- + +def print_report(report: DiagnosticReport) -> None: + """Print a colour-coded diagnostic report to the terminal.""" + RESET = "\033[0m" + GREEN = "\033[92m" + YELLOW = "\033[93m" + RED = "\033[91m" + BOLD = "\033[1m" + + dc = report.domain_config + print(f"\n{BOLD}╔══════════════════════════════════════════════════════╗{RESET}") + print(f"{BOLD}║ Deployment Doctor — {dc.domain:<33}║{RESET}") + print(f"{BOLD}╚══════════════════════════════════════════════════════╝{RESET}\n") + + status_icons = { + DiagnosticStatus.OK: f"{GREEN}✔{RESET}", + DiagnosticStatus.WARNING: f"{YELLOW}⚠{RESET}", + DiagnosticStatus.CRITICAL: f"{RED}✘{RESET}", + } + + for check in report.checks: + icon = status_icons.get(check.status, "?") + print(f" {icon} {BOLD}{check.name}{RESET}: {check.message}") + if check.suggestion: + print(f" └─ suggestion: {check.suggestion}") + + print() + if report.is_healthy: + print(f" {GREEN}{BOLD}All checks passed! Your deployment is healthy.{RESET}\n") + else: + c = report.critical_count + w = report.warning_count + parts = [] + if c: + parts.append(f"{RED}{c} critical{RESET}") + if w: + parts.append(f"{YELLOW}{w} warning(s){RESET}") + print(f" Summary: {', '.join(parts)}\n") diff --git a/core/executor.py b/core/executor.py new file mode 100644 index 0000000..bccfdd6 --- /dev/null +++ b/core/executor.py @@ -0,0 +1,77 @@ +# -*- coding: utf-8 -*- +""" +System command executor with structured error handling. + +**Single Responsibility:** Execute OS-level commands and manage files — +nothing more. Provides a ``DryRunExecutor`` substitute for safe testing +without side effects (Liskov Substitution Principle). +""" + +from __future__ import annotations + +import shutil +import subprocess +from datetime import datetime +from pathlib import Path +from typing import Optional, Tuple + +from core.interfaces import ISystemExecutor + + +class LinuxExecutor(ISystemExecutor): + """Production executor — runs real commands on the host system.""" + + def run(self, command: str, check: bool = True) -> Tuple[int, str, str]: + """Execute *command* via ``/bin/bash`` and return (rc, stdout, stderr).""" + result = subprocess.run( + command, + shell=True, + executable="/bin/bash", + capture_output=True, + text=True, + ) + if check and result.returncode != 0: + raise RuntimeError( + f"Command failed (rc={result.returncode}): {command}\n" + f"stderr: {result.stderr.strip()}" + ) + return result.returncode, result.stdout.strip(), result.stderr.strip() + + def write_file(self, path: Path, content: str, mode: int = 0o644) -> None: + """Write *content* to *path* with the given POSIX *mode*.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + path.chmod(mode) + + def file_exists(self, path: Path) -> bool: + return path.exists() + + def backup_file(self, path: Path) -> Optional[Path]: + """Create a timestamped backup of *path*.""" + if not path.exists(): + return None + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + backup_path = path.with_suffix(f".bak.{timestamp}") + shutil.copy2(str(path), str(backup_path)) + return backup_path + + +class DryRunExecutor(ISystemExecutor): + """Test / dry-run executor — logs actions without side effects (LSP).""" + + def __init__(self) -> None: + self.log: list[str] = [] + + def run(self, command: str, check: bool = True) -> Tuple[int, str, str]: + self.log.append(f"[DRY-RUN] exec: {command}") + return 0, "", "" + + def write_file(self, path: Path, content: str, mode: int = 0o644) -> None: + self.log.append(f"[DRY-RUN] write: {path} (mode={oct(mode)}, {len(content)} bytes)") + + def file_exists(self, path: Path) -> bool: + return path.exists() + + def backup_file(self, path: Path) -> Optional[Path]: + self.log.append(f"[DRY-RUN] backup: {path}") + return None diff --git a/core/interfaces.py b/core/interfaces.py new file mode 100644 index 0000000..224f3f6 --- /dev/null +++ b/core/interfaces.py @@ -0,0 +1,89 @@ +# -*- coding: utf-8 -*- +""" +Abstract interfaces (contracts) for the deployment engine. + +Applying the **Interface Segregation Principle** (ISP) and +**Dependency Inversion Principle** (DIP): every high-level orchestrator +depends on these thin abstractions — never on concrete implementations. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from pathlib import Path +from typing import Optional, Tuple + +from core.models import ( + DiagnosticReport, + DjangoProjectInfo, + DomainConfig, + ServiceConfig, +) + + +class IDjangoDetector(ABC): + """Inspects a Django project directory and returns structured metadata.""" + + @abstractmethod + def detect(self, web_dir: Path, project_hint: Optional[str] = None) -> DjangoProjectInfo: + """Analyse *web_dir* and return a populated ``DjangoProjectInfo``.""" + + +class IServiceGenerator(ABC): + """Renders systemd service unit content from a ``ServiceConfig``.""" + + @abstractmethod + def generate(self, config: ServiceConfig) -> str: + """Return the full text of a systemd ``.service`` unit file.""" + + +class ISystemExecutor(ABC): + """Abstraction over OS-level command execution.""" + + @abstractmethod + def run(self, command: str, check: bool = True) -> Tuple[int, str, str]: + """Execute *command* and return ``(return_code, stdout, stderr)``.""" + + @abstractmethod + def write_file(self, path: Path, content: str, mode: int = 0o644) -> None: + """Write *content* to *path* with the given POSIX *mode*.""" + + @abstractmethod + def file_exists(self, path: Path) -> bool: + """Return ``True`` if *path* exists on the filesystem.""" + + @abstractmethod + def backup_file(self, path: Path) -> Optional[Path]: + """Create a timestamped backup of *path* and return the backup path.""" + + +class INginxManager(ABC): + """Manages HestiaCP NGINX proxy template installation.""" + + @abstractmethod + def install_templates(self, tpl_source: Path, stpl_source: Path) -> None: + """Copy templates to the HestiaCP template directory and set permissions.""" + + @abstractmethod + def rebuild_domain(self, domain_config: DomainConfig) -> None: + """Force-rebuild the NGINX configuration for a domain.""" + + +class IProjectMigrator(ABC): + """Upgrades / repairs a previously deployed project in-place.""" + + @abstractmethod + def needs_migration(self, domain_config: DomainConfig) -> bool: + """Return ``True`` if the existing deployment requires migration.""" + + @abstractmethod + def migrate(self, domain_config: DomainConfig) -> DiagnosticReport: + """Perform the migration / repair and return a diagnostic report.""" + + +class IDeploymentDoctor(ABC): + """Runs post-deployment health checks.""" + + @abstractmethod + def diagnose(self, domain_config: DomainConfig) -> DiagnosticReport: + """Run all health checks and return a ``DiagnosticReport``.""" diff --git a/core/migrator.py b/core/migrator.py new file mode 100644 index 0000000..09c4e1d --- /dev/null +++ b/core/migrator.py @@ -0,0 +1,200 @@ +# -*- coding: utf-8 -*- +""" +Project migrator — upgrades and repairs previously deployed projects. + +**Single Responsibility:** Scan an existing systemd service file, detect +misconfigurations (wrong socket path, missing ``--umask``, incorrect WSGI +module), and apply corrective changes without downtime where possible. + +""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import Optional + +from core.detector import DjangoProjectDetector +from core.interfaces import IProjectMigrator, IServiceGenerator, ISystemExecutor +from core.models import ( + DiagnosticCheck, + DiagnosticReport, + DiagnosticStatus, + DjangoProjectInfo, + DomainConfig, + ServiceConfig, +) + + +class ProjectMigrator(IProjectMigrator): + """Scans and repairs previously deployed Gunicorn services.""" + + def __init__( + self, + executor: ISystemExecutor, + detector: DjangoProjectDetector, + generator: IServiceGenerator, + ) -> None: + self._executor = executor + self._detector = detector + self._generator = generator + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def needs_migration(self, domain_config: DomainConfig) -> bool: + """Return ``True`` if the existing deployment has known issues.""" + service_file = domain_config.service_file + if not self._executor.file_exists(service_file): + return False + + content = service_file.read_text(encoding="utf-8", errors="replace") + + checks = [ + "--umask" not in content, + self._has_socket_mismatch(content, domain_config), + self._has_wsgi_mismatch(content, domain_config), + "ProtectSystem" not in content, + ] + return any(checks) + + def migrate(self, domain_config: DomainConfig) -> DiagnosticReport: + """Perform the migration / repair and return a diagnostic report.""" + report = DiagnosticReport(domain_config=domain_config) + dc = domain_config + + # --- Step 1: Read existing service file --- + if not self._executor.file_exists(dc.service_file): + report.checks.append(DiagnosticCheck( + name="Service File", + status=DiagnosticStatus.CRITICAL, + message=f"Service file not found: {dc.service_file}", + suggestion="Run 'setup' command first to create the service.", + )) + return report + + # --- Step 2: Backup existing service file --- + backup_path = self._executor.backup_file(dc.service_file) + report.checks.append(DiagnosticCheck( + name="Backup", + status=DiagnosticStatus.OK, + message=f"Backed up to: {backup_path}", + )) + + # --- Step 3: Re-detect the project (smart detection) --- + project_info = self._detector.detect(dc.web_dir) + if not project_info.is_valid: + report.checks.append(DiagnosticCheck( + name="Project Detection", + status=DiagnosticStatus.CRITICAL, + message="Could not detect project structure.", + suggestion=f"Errors: {'; '.join(project_info.errors)}", + )) + return report + + report.checks.append(DiagnosticCheck( + name="Project Detection", + status=DiagnosticStatus.OK, + message=f"Layout: {project_info.layout.name}, " + f"WSGI: {project_info.wsgi_module}", + )) + + # --- Step 4: Generate corrected service file --- + service_config = ServiceConfig( + domain_config=dc, + project_info=project_info, + ) + new_content = self._generator.generate(service_config) + self._executor.write_file(dc.service_file, new_content, mode=0o644) + + report.checks.append(DiagnosticCheck( + name="Service File Update", + status=DiagnosticStatus.OK, + message="Service file regenerated with --umask 007, hardened security, " + f"and correct WSGI module ({project_info.wsgi_module}).", + )) + + # --- Step 5: Fix file permissions --- + self._fix_permissions(dc, project_info, report) + + # --- Step 6: Reload and restart --- + self._restart_service(dc, report) + + return report + + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ + + def _has_socket_mismatch(self, content: str, dc: DomainConfig) -> bool: + """Check if the socket path in the service file differs from canonical.""" + match = re.search(r"unix:(\S+)", content) + if match: + existing = Path(match.group(1)) + return existing != dc.socket_path + return True + + def _has_wsgi_mismatch(self, content: str, dc: DomainConfig) -> bool: + """Check if the WSGI module might be wrong for the current project.""" + project_info = self._detector.detect(dc.web_dir) + if not project_info.is_valid: + return False + return project_info.wsgi_module not in content + + def _fix_permissions( + self, + dc: DomainConfig, + project_info: DjangoProjectInfo, + report: DiagnosticReport, + ) -> None: + """Fix ownership and permissions on db.sqlite3 and static directory.""" + web_dir = dc.web_dir + + # Fix db.sqlite3 + db_file = web_dir / "db.sqlite3" + if self._executor.file_exists(db_file): + self._executor.run( + f"chown {dc.user}:www-data {db_file} && chmod 664 {db_file}", + check=False, + ) + report.checks.append(DiagnosticCheck( + name="Database Permissions", + status=DiagnosticStatus.OK, + message=f"Set {dc.user}:www-data 664 on db.sqlite3", + )) + + # Fix static root directory + static_dir = web_dir / project_info.static_root_name + if self._executor.file_exists(static_dir): + self._executor.run( + f"chown -R {dc.user}:www-data {static_dir}", + check=False, + ) + report.checks.append(DiagnosticCheck( + name="Static Files Permissions", + status=DiagnosticStatus.OK, + message=f"Fixed ownership on {static_dir}", + )) + + def _restart_service(self, dc: DomainConfig, report: DiagnosticReport) -> None: + """Reload systemd daemon and restart the Gunicorn service.""" + self._executor.run("systemctl daemon-reload", check=False) + rc, _, stderr = self._executor.run( + f"systemctl restart {dc.service_name}", check=False + ) + + if rc == 0: + report.checks.append(DiagnosticCheck( + name="Service Restart", + status=DiagnosticStatus.OK, + message=f"{dc.service_name} restarted successfully.", + )) + else: + report.checks.append(DiagnosticCheck( + name="Service Restart", + status=DiagnosticStatus.CRITICAL, + message=f"Failed to restart {dc.service_name}.", + suggestion=f"Check logs: journalctl -u {dc.service_name} -n 30\n" + f"Error: {stderr}", + )) diff --git a/core/models.py b/core/models.py new file mode 100644 index 0000000..98e151e --- /dev/null +++ b/core/models.py @@ -0,0 +1,184 @@ +# -*- coding: utf-8 -*- +""" +Domain models and value objects for the HestiaCP Django-Gunicorn deployment engine. + +These dataclasses serve as the **Single Source of Truth** for all path +conventions, service naming rules, and configuration parameters used across +every module. Centralising them here eliminates magic strings and satisfies +the DRY principle. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from enum import Enum, auto +from pathlib import Path +from typing import List, Optional + + +# --------------------------------------------------------------------------- +# Enumerations +# --------------------------------------------------------------------------- + +class ProjectLayout(Enum): + """Detected Django project directory layout.""" + + MODULAR = auto() # config/wsgi.py (clean / modern) + TRADITIONAL = auto() # /wsgi.py (django-admin startproject) + FLAT = auto() # wsgi.py at project root (rare but valid) + UNKNOWN = auto() + + +class DatabaseEngine(Enum): + """Database back-end detected in settings.""" + + SQLITE = auto() + POSTGRESQL = auto() + MYSQL = auto() + OTHER = auto() + UNKNOWN = auto() + + +class DiagnosticStatus(Enum): + """Result status for a single diagnostic check.""" + + OK = auto() + WARNING = auto() + CRITICAL = auto() + + +# --------------------------------------------------------------------------- +# Path & naming conventions (Single Source of Truth) +# --------------------------------------------------------------------------- + +HESTIA_TEMPLATE_DIR = Path("/usr/local/hestia/data/templates/web/nginx") +SYSTEMD_SERVICE_DIR = Path("/etc/systemd/system") +TEMPLATE_NAME = "django-gunicorn" +SOCKET_FILENAME = "gunicorn.sock" +DEFAULT_VENV_CANDIDATES = ("venv", ".venv", "env") +DEFAULT_STATIC_DIR = "staticfiles" + + +def build_home_dir(user: str) -> Path: + """Return the HestiaCP home directory for *user*.""" + return Path(f"/home/{user}") + + +def build_web_dir(user: str, domain: str) -> Path: + """Return the web-root (``public_html``) for a domain.""" + return build_home_dir(user) / "web" / domain / "public_html" + + +def build_socket_path(user: str, domain: str) -> Path: + """Return the canonical unix-socket path.""" + return build_web_dir(user, domain) / SOCKET_FILENAME + + +def build_service_name(domain: str) -> str: + """Return the systemd service unit name (without ``.service``).""" + return f"gunicorn-{domain}" + + +def build_service_file(domain: str) -> Path: + """Return the absolute path to the systemd unit file.""" + return SYSTEMD_SERVICE_DIR / f"{build_service_name(domain)}.service" + + +# --------------------------------------------------------------------------- +# Core data-transfer objects +# --------------------------------------------------------------------------- + +@dataclass(frozen=True) +class DomainConfig: + """Immutable value object describing a HestiaCP domain deployment target.""" + + domain: str + user: str + + @property + def home_dir(self) -> Path: + return build_home_dir(self.user) + + @property + def web_dir(self) -> Path: + return build_web_dir(self.user, self.domain) + + @property + def socket_path(self) -> Path: + return build_socket_path(self.user, self.domain) + + @property + def service_name(self) -> str: + return build_service_name(self.domain) + + @property + def service_file(self) -> Path: + return build_service_file(self.domain) + + +@dataclass +class DjangoProjectInfo: + """Result of inspecting the Django project inside ``public_html``.""" + + layout: ProjectLayout = ProjectLayout.UNKNOWN + wsgi_module: str = "" + venv_path: Optional[Path] = None + settings_module: str = "" + static_root_name: str = DEFAULT_STATIC_DIR + database_engine: DatabaseEngine = DatabaseEngine.UNKNOWN + has_dot_env: bool = False + has_requirements: bool = False + errors: List[str] = field(default_factory=list) + + @property + def is_valid(self) -> bool: + return self.layout != ProjectLayout.UNKNOWN and bool(self.wsgi_module) + + +@dataclass +class ServiceConfig: + """All parameters needed to render a systemd unit file.""" + + domain_config: DomainConfig + project_info: DjangoProjectInfo + workers: int = 0 + threads: int = 2 + umask: str = "007" + log_level: str = "info" + restart_sec: int = 3 + + def __post_init__(self) -> None: + if self.workers <= 0: + cpu = os.cpu_count() or 1 + self.workers = (2 * cpu) + 1 + + +@dataclass +class DiagnosticCheck: + """A single diagnostic check result.""" + + name: str + status: DiagnosticStatus + message: str + suggestion: str = "" + + +@dataclass +class DiagnosticReport: + """Collection of diagnostic checks for a deployment.""" + + domain_config: DomainConfig + checks: List[DiagnosticCheck] = field(default_factory=list) + + @property + def is_healthy(self) -> bool: + return all(c.status == DiagnosticStatus.OK for c in self.checks) + + @property + def critical_count(self) -> int: + return sum(1 for c in self.checks if c.status == DiagnosticStatus.CRITICAL) + + @property + def warning_count(self) -> int: + return sum(1 for c in self.checks if c.status == DiagnosticStatus.WARNING) diff --git a/core/nginx_manager.py b/core/nginx_manager.py new file mode 100644 index 0000000..17f3ce0 --- /dev/null +++ b/core/nginx_manager.py @@ -0,0 +1,59 @@ +# -*- coding: utf-8 -*- +""" +NGINX template manager for HestiaCP. + +**Single Responsibility:** Install, validate, and manage the custom NGINX +proxy templates that HestiaCP uses to route traffic to the Gunicorn socket. +""" + +from __future__ import annotations + +from pathlib import Path + +from core.interfaces import INginxManager, ISystemExecutor +from core.models import HESTIA_TEMPLATE_DIR, TEMPLATE_NAME, DomainConfig + + +class HestiaNginxManager(INginxManager): + """Manages NGINX template deployment for HestiaCP.""" + + def __init__(self, executor: ISystemExecutor) -> None: + self._executor = executor + + @property + def tpl_dest(self) -> Path: + return HESTIA_TEMPLATE_DIR / f"{TEMPLATE_NAME}.tpl" + + @property + def stpl_dest(self) -> Path: + return HESTIA_TEMPLATE_DIR / f"{TEMPLATE_NAME}.stpl" + + def install_templates(self, tpl_source: Path, stpl_source: Path) -> None: + """Copy templates to the HestiaCP template directory and set permissions.""" + for src, dst in [(tpl_source, self.tpl_dest), (stpl_source, self.stpl_dest)]: + if not self._executor.file_exists(src): + raise FileNotFoundError(f"Template source not found: {src}") + + content = src.read_text(encoding="utf-8") + self._executor.write_file(dst, content, mode=0o644) + + # Set ownership to root:root + self._executor.run( + f"chown root:root {self.tpl_dest} {self.stpl_dest}" + ) + + def rebuild_domain(self, domain_config: DomainConfig) -> None: + """Force-rebuild the NGINX configuration for a domain and restart.""" + dc = domain_config + self._executor.run( + f"v-rebuild-web-domain {dc.user} {dc.domain} yes", + check=False, + ) + self._executor.run("systemctl restart nginx", check=False) + + def is_template_installed(self) -> bool: + """Check if both template files exist in the HestiaCP directory.""" + return ( + self._executor.file_exists(self.tpl_dest) + and self._executor.file_exists(self.stpl_dest) + ) diff --git a/core/service_generator.py b/core/service_generator.py new file mode 100644 index 0000000..109e3a0 --- /dev/null +++ b/core/service_generator.py @@ -0,0 +1,84 @@ +# -*- coding: utf-8 -*- +""" +Systemd service unit file generator for Gunicorn. + +**Single Responsibility:** Accept a ``ServiceConfig`` and render a production- +hardened systemd ``.service`` file. Security best-practices (``--umask 007``, +``ProtectSystem``, dynamic worker count) are baked in by default. +""" + +from __future__ import annotations + +import textwrap + +from core.interfaces import IServiceGenerator +from core.models import ServiceConfig + + +class SystemdServiceGenerator(IServiceGenerator): + """Generates a systemd service unit for a Gunicorn-backed Django app.""" + + _TEMPLATE = textwrap.dedent("""\ + # ────────────────────────────────────────────────────────────────── + # Gunicorn service for {domain} + # Auto-generated by hestia-django-gunicorn deployment engine + # ────────────────────────────────────────────────────────────────── + [Unit] + Description=Gunicorn instance for {domain} (Django) + After=network.target + + [Service] + User={user} + Group=www-data + WorkingDirectory={web_dir} + Environment="PATH={venv_bin}:/usr/local/bin:/usr/bin" + Environment="PYTHONUNBUFFERED=1" + {env_file_line} + ExecStart={venv_bin}/gunicorn \\ + --access-logfile - \\ + --error-logfile - \\ + --workers {workers} \\ + --threads {threads} \\ + --worker-class=gthread \\ + --bind unix:{socket_path} \\ + --umask {umask} \\ + --log-level={log_level} \\ + {wsgi_module} + + Restart=always + RestartSec={restart_sec} + ProtectSystem=full + PrivateTmp=true + NoNewPrivileges=true + + [Install] + WantedBy=multi-user.target + """) + + def generate(self, config: ServiceConfig) -> str: + """Return the full text of a systemd ``.service`` unit file.""" + dc = config.domain_config + pi = config.project_info + + venv_bin = str(pi.venv_path / "bin") if pi.venv_path else "/usr/bin" + + # Optional EnvironmentFile directive for projects with .env + if pi.has_dot_env: + env_file_line = f"EnvironmentFile={dc.web_dir}/.env" + else: + env_file_line = "# No .env found — environment variables from Environment= only" + + return self._TEMPLATE.format( + domain=dc.domain, + user=dc.user, + web_dir=dc.web_dir, + venv_bin=venv_bin, + socket_path=dc.socket_path, + workers=config.workers, + threads=config.threads, + umask=config.umask, + log_level=config.log_level, + restart_sec=config.restart_sec, + wsgi_module=pi.wsgi_module, + env_file_line=env_file_line, + )