import os import subprocess from datetime import datetime class DeploymentController: def __init__(self, logger): self.logger = logger self.ssh_key = None self.config_dir = None self.remote_user = "ubuntu" self.remote_host = "3.9.182.122" self.certbot_email = "azeem.fidahusein@gmail.com" self.remote_temp_dir = "nginx_deploy_temp" self.dest_nginx_path = "/etc/nginx/" self.dest_sites_path = "/etc/nginx/sites-available/" self.source_nginx_conf = "nginx.conf" self.source_sites_dir = "sites-available" self.domains = [] self.config_files = [] def preflight_checks(self): self.logger.log("Running pre-flight checks...") if not self.ssh_key or not os.path.isfile(self.ssh_key): self.logger.log("ERROR: SSH key not set or not found.") return False if not self.config_dir or not os.path.isdir(self.config_dir): self.logger.log("ERROR: Config directory not set or not found.") return False nginx_conf = os.path.join(self.config_dir, self.source_nginx_conf) sites_dir = os.path.join(self.config_dir, self.source_sites_dir) if not os.path.isfile(nginx_conf): self.logger.log(f"ERROR: nginx.conf not found in {self.config_dir}.") return False if not os.path.isdir(sites_dir): self.logger.log(f"ERROR: sites-available not found in {self.config_dir}.") return False self.logger.log("Pre-flight checks passed.") return True def scan_domains(self): self.logger.log("Scanning for domains in configs...") sites_dir = os.path.join(self.config_dir, self.source_sites_dir) self.config_files = [f for f in os.listdir(sites_dir) if os.path.isfile(os.path.join(sites_dir, f))] domains = set() for fname in self.config_files: path = os.path.join(sites_dir, fname) with open(path, 'r') as f: for line in f: if 'server_name' in line: parts = line.split() if 'server_name' in parts: idx = parts.index('server_name') found = parts[idx+1:] if idx+1 < len(parts) else [] for d in found: d = d.strip(';') if d: domains.add(d) self.domains = sorted(domains) if self.domains: self.logger.log(f"Found domains: {' '.join(self.domains)}") else: self.logger.log("WARNING: No domains found.") return self.domains def transfer_files(self): self.logger.log("Transferring files to remote server...") if not self.preflight_checks(): return False ssh_key = self.ssh_key temp_dir = self.remote_temp_dir # Create temp dir on remote cmd1 = ["ssh", "-i", ssh_key, f"{self.remote_user}@{self.remote_host}", f"mkdir -p {temp_dir}"] self._run_cmd(cmd1, "Create remote temp dir") # Copy nginx.conf nginx_conf = os.path.join(self.config_dir, self.source_nginx_conf) cmd2 = ["scp", "-i", ssh_key, nginx_conf, f"{self.remote_user}@{self.remote_host}:{temp_dir}/"] self._run_cmd(cmd2, "Copy nginx.conf") # Copy sites-available sites_dir = os.path.join(self.config_dir, self.source_sites_dir) cmd3 = ["scp", "-i", ssh_key, "-r", sites_dir, f"{self.remote_user}@{self.remote_host}:{temp_dir}/"] self._run_cmd(cmd3, "Copy sites-available dir") self.logger.log("File transfer complete.") return True def remote_ops(self): self.logger.log("Moving transferred files to final destination, ensuring symlinks, and reloading Nginx...") ssh_key = self.ssh_key temp_dir = self.remote_temp_dir config_files = ' '.join(self.config_files) remote_script = f''' sudo mv {temp_dir}/nginx.conf {self.dest_nginx_path}nginx.conf sudo mv {temp_dir}/sites-available/* {self.dest_sites_path} for CONFIG_FILE in {config_files} do SOURCE_FILE="/etc/nginx/sites-available/$CONFIG_FILE" LINK_FILE="/etc/nginx/sites-enabled/$CONFIG_FILE" if [ ! -L "$LINK_FILE" ]; then if [ -f "$SOURCE_FILE" ]; then sudo ln -s "$SOURCE_FILE" "$LINK_FILE" fi fi done sudo nginx -t && sudo systemctl reload nginx ''' cmd = ["ssh", "-i", ssh_key, f"{self.remote_user}@{self.remote_host}", remote_script] self._run_cmd(cmd, "Move files, ensure symlinks, and reload Nginx") self.logger.log("Remote file move, symlink check, and reload complete.") return True def run_certbot(self, prompt_fn=None): if not self.domains: self.logger.log("No domains found, skipping Certbot.") return False if prompt_fn: proceed = prompt_fn("Run Certbot for the discovered domains? (y/n)") if not proceed: self.logger.log("Certbot step skipped by user.") return False self.logger.log("Running Certbot on remote server...") ssh_key = self.ssh_key domains_args = ' '.join([f"-d {d}" for d in self.domains]) certbot_cmd = f"sudo certbot --nginx --non-interactive --agree-tos --email {self.certbot_email} --redirect --expand {domains_args}" cmd = ["ssh", "-t", "-i", ssh_key, f"{self.remote_user}@{self.remote_host}", certbot_cmd] self._run_cmd(cmd, "Run Certbot") self.logger.log("Certbot step complete.") return True def cleanup(self): self.logger.log("Performing cleanup...") ssh_key = self.ssh_key temp_dir = self.remote_temp_dir cmd = ["ssh", "-i", ssh_key, f"{self.remote_user}@{self.remote_host}", f"rm -rf {temp_dir}"] self._run_cmd(cmd, "Cleanup remote temp dir") self.logger.log("Cleanup complete.") return True def _run_cmd(self, cmd, desc): self.logger.log(f"[CMD] {desc}: {' '.join(cmd)}") try: proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) out, err = proc.communicate() if out: self.logger.log(out.strip()) if err: self.logger.log(err.strip()) if proc.returncode != 0: self.logger.log(f"ERROR: Command failed with code {proc.returncode}") except Exception as e: self.logger.log(f"Exception running command: {e}")