#!/usr/bin/env python3 """SSH worker, sent over stdin. Never installed or executed automatically.""" import base64 import fcntl import gzip import hashlib import json import os from pathlib import Path import pwd import re import shutil import signal import subprocess import sys import tarfile import time import urllib.request def digest(path): value = hashlib.sha256() with open(path, 'rb') as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b''): value.update(chunk) return value.hexdigest() def validate_config(config): for key in ['app_dir', 'deploy_dir', 'node_bin', 'pm2_home', 'database_dump_bin']: value = config[key] if not re.fullmatch(r'/[A-Za-z0-9_./-]+', value) or '..' in Path(value).parts: raise ValueError(f'Invalid absolute path for {key}') backup_script = config.get('database_backup_script') if backup_script and (not re.fullmatch(r'/[A-Za-z0-9_./-]+', backup_script) or '..' in Path(backup_script).parts): raise ValueError('Invalid database backup script path') app, work = Path(config['app_dir']).resolve(), Path(config['deploy_dir']).resolve() if len(app.parts) < 4 or len(work.parts) < 4 or app == work or app in work.parents or work in app.parents: raise ValueError('Application and deployment directories must be separate specific paths') if Path(config['app_dir']).is_symlink(): raise ValueError('This workflow expects the existing real application directory') if not re.fullmatch(r'[A-Za-z0-9_-]+', config['pm2_app']) or config['pm2_app'] == 'all': raise ValueError('Configure a single named PM2 app') protected = config['protected'] if not {'.env', 'uploads', 'cert.pem', 'key.pem'} <= set(protected): raise ValueError('Protect the server environment, uploads and certificate keys') if any('/' in item or item in {'', '.', '..', 'node_modules', 'server.js'} for item in protected): raise ValueError('Protected paths must be top-level persistent entries') for key in ['health_url', 'public_health_url', 'rollback_health_url', 'rollback_public_health_url']: if not config[key].startswith(('http://', 'https://')): raise ValueError(f'Invalid {key}') return app, work def validate_release(path, protected): with tarfile.open(path) as archive: names = set() for member in archive.getmembers(): parts = Path(member.name).parts if not parts or Path(member.name).is_absolute() or '..' in parts: raise ValueError('Unsafe archive path') if not (member.isfile() or member.isdir()) or parts[0] in set(protected) | {'node_modules', '.git', '.deploy'}: raise ValueError('Unsafe archive entry') names.add(member.name) if not {'package.json', 'package-lock.json', 'server.js'} <= names: raise ValueError('Incomplete release archive') return sum(member.size for member in archive.getmembers()) def directory_size(path): return sum(item.stat().st_size for item in path.rglob('*') if item.is_file() and not item.is_symlink()) def daemon_running(config): pid = int((Path(config['pm2_home']) / 'pm2.pid').read_text().strip()) os.kill(pid, 0) if 'PM2' not in Path('/proc', str(pid), 'comm').read_text(): raise ValueError('Configured PM2 PID is not a PM2 daemon') if not (Path(config['pm2_home']) / 'rpc.sock').is_socket(): raise ValueError('Configured PM2 daemon socket unavailable') def inspect(config): app, work = validate_config(config) daemon_running(config) saved = json.loads((Path(config['pm2_home']) / 'dump.pm2').read_text()) applications = [{key: row.get(key) for key in ['name', 'pm_cwd', 'pm_exec_path', 'exec_mode', 'uid']} for row in saved] source = {} source_dirs = {'controllers', 'middlewares', 'routes', 'utils', 'services', 'migrations', 'scripts'} for file in app.rglob('*'): relative = file.relative_to(app) if file.is_file() and not file.is_symlink() and ( (len(relative.parts) == 1 and (file.suffix == '.js' or file.name in {'package.json', 'package-lock.json'})) or (relative.parts[0] in source_dirs and file.suffix in {'.js', '.sql', '.py', '.sh'})): source[str(relative)] = digest(file) print(json.dumps({'app_dir': str(app), 'env_exists': (app / '.env').is_file(), 'uploads_exists': (app / 'uploads').is_dir(), 'git_checkout': (app / '.git').exists(), 'pm2_daemon_running': True, 'saved_apps': applications, 'database_backup_mode': 'external' if config.get('database_backup_script') else 'bundled', 'database_dump_available': os.access(config['database_dump_bin'], os.X_OK), 'app_size_bytes': directory_size(app), 'free_bytes': shutil.disk_usage(app).free, 'deploy_dir': str(work), 'source_checksums': source}, indent=2)) class Deployment: def __init__(self, config, release): self.config = config self.app, self.work = validate_config(config) if not re.fullmatch(r'[0-9a-f]{40}', release): raise ValueError('Invalid release identifier') self.release = release self.backup = self.work / 'backups' / release self.stage = self.work / 'releases' / release self.env = {**os.environ, 'PATH': config['node_bin'] + ':' + os.environ.get('PATH', ''), 'PM2_HOME': config['pm2_home']} self.pm2 = str(Path(config['node_bin']) / 'pm2') self.log = None def run(self, args, **kwargs): environment = kwargs.pop('env', self.env) return subprocess.run(args, env=environment, check=True, stdout=self.log or subprocess.DEVNULL, stderr=self.log or subprocess.DEVNULL, **kwargs) def verify_target(self, require_online=True): daemon_running(self.config) result = subprocess.run([self.pm2, 'jlist'], env=self.env, check=True, capture_output=True, text=True) rows = json.loads(result.stdout) targets = [row for row in rows if row['name'] == self.config['pm2_app']] if len(targets) != 1: raise ValueError('Expected exactly one existing PM2 target') row = targets[0] env = row['pm2_env'] if Path(env['pm_cwd']).resolve() != self.app or Path(env['pm_exec_path']).resolve() != self.app / 'server.js': raise ValueError('PM2 target does not match the reviewed application path') if env.get('watch') or env.get('exec_mode') != 'fork_mode': raise ValueError('Disable watch and review non-fork processes before using this workflow') pid = row.get('pid', 0) if require_online and (not pid or env.get('status') != 'online'): raise ValueError('Existing app must be online before deployment') expected_uid = pwd.getpwnam(self.config['app_user']).pw_uid actual_uid = Path('/proc', str(pid)).stat().st_uid if pid else int(env.get('uid', -1)) if actual_uid != expected_uid: raise ValueError('Live application user differs from configuration') return rows def sync(self, source): self.run(['rsync', '-a', '--delete', *[f'--exclude=/{name}' for name in self.config['protected']], str(source) + '/', str(self.app) + '/']) def health(self, rollback=False): keys = ['rollback_health_url', 'rollback_public_health_url'] if rollback else ['health_url', 'public_health_url'] for attempt in range(15): try: for key in keys: with urllib.request.urlopen(self.config[key], timeout=5) as response: if response.status != 200: raise ValueError('HTTP status is not 200') if not rollback: data = json.loads(response.read(65536)) if (data.get('success') is not True or data.get('service') != 'bondvia-backend' or data.get('commit') != self.release): raise ValueError('Unexpected health response') return except (OSError, ValueError): time.sleep(2) raise ValueError('Health checks failed') def backup_app(self, processes): self.backup.mkdir(mode=0o700) (self.backup / 'pm2.json').write_text(json.dumps(processes, indent=2)) self.run(['tar', '-czf', str(self.backup / 'app.tar.gz'), '-C', str(self.app), '.']) if self.config.get('database_backup_script'): self.run(['bash', self.config['database_backup_script'], str(self.backup / 'database.sql.gz')]) else: self.run(['node', str(self.stage / 'scripts' / 'backup-db.js'), str(self.app), str(self.backup / 'database.sql.gz'), self.config['database_dump_bin'], str(self.backup / 'pm2.json'), self.config['pm2_app']]) for name in ['app.tar.gz', 'database.sql.gz']: with gzip.open(self.backup / name, 'rb') as stream: while stream.read(1024 * 1024): pass if (self.backup / 'database.sql.gz').stat().st_size < 100: raise ValueError('Database backup is empty or unexpectedly small') with tarfile.open(self.backup / 'app.tar.gz') as archive: if not any(member.name.lstrip('./') == 'server.js' for member in archive.getmembers()): raise ValueError('Application backup missing server.js') checksums = {name: digest(self.backup / name) for name in ['app.tar.gz', 'database.sql.gz', 'pm2.json']} manifest = {'release': self.release, 'app_dir': str(self.app), 'pm2_app': self.config['pm2_app'], 'created_utc': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()), 'checksums': checksums} (self.backup / 'manifest.json').write_text(json.dumps(manifest, indent=2)) print(f'Backup verified: {self.backup}', flush=True) def verify_backup(self): manifest = json.loads((self.backup / 'manifest.json').read_text()) if manifest['app_dir'] != str(self.app) or manifest['pm2_app'] != self.config['pm2_app']: raise ValueError('Backup belongs to a different application') if digest(self.backup / 'app.tar.gz') != manifest['checksums']['app.tar.gz']: raise ValueError('Backup checksum mismatch') restore = self.work / 'restore' / self.release if restore.exists(): raise ValueError('Restore directory already exists; inspect interrupted rollback') with tarfile.open(self.backup / 'app.tar.gz') as archive: required = 2 * sum(member.size for member in archive.getmembers()) if shutil.disk_usage(self.work).free < required + int(self.config['minimum_free_bytes']): raise ValueError('Insufficient space to restore the backup safely') return restore def restore_code(self): restore = self.verify_backup() restore.mkdir(parents=True, mode=0o700) self.run(['tar', '-xzf', str(self.backup / 'app.tar.gz'), '-C', str(restore)]) self.sync(restore) self.run([self.pm2, 'restart', self.config['pm2_app']]) self.health(rollback=True) print('Previous code and dependencies restored; database and live uploads were preserved.', flush=True) def deploy(self, checksum, migrate): archive = self.work / 'incoming' / f'{self.release}.tar' if digest(archive) != checksum: raise ValueError('Uploaded release checksum mismatch') expanded = validate_release(archive, self.config['protected']) if self.backup.exists() or self.stage.exists(): raise ValueError('Release already attempted; inspect it before retrying with a new commit') for name in ['rsync', 'tar', 'gzip', 'bash']: if not shutil.which(name): raise ValueError(f'Missing command: {name}') if not (self.app / '.env').is_file() or not (self.app / 'uploads').is_dir(): raise ValueError('Production .env or uploads directory missing') if self.config.get('database_backup_script') and not Path(self.config['database_backup_script']).is_file(): raise ValueError('Configured external database backup script is missing') if not self.config.get('database_backup_script') and not os.access(self.config['database_dump_bin'], os.X_OK): raise ValueError('Database dump utility missing') needed = 2 * directory_size(self.app) + 2 * expanded + int(self.config['database_size_reserve_bytes']) if shutil.disk_usage(self.work).free < needed + int(self.config['minimum_free_bytes']): raise ValueError('Insufficient disk space for staging, full backup, and rollback') processes = self.verify_target() self.health(rollback=True) self.stage.mkdir(parents=True, mode=0o755) with tarfile.open(archive) as stream: stream.extractall(self.stage, filter='data') if not self.config.get('database_backup_script') and not (self.stage / 'scripts' / 'backup-db.js').is_file(): raise ValueError('Release must include the bundled database backup helper') if (self.stage / 'migrations').exists() and not migrate: raise ValueError('Release has migrations: review them and explicitly select --migrate') self.log = open(self.work / f'{self.release}.log', 'a', buffering=1) try: self.run(['npm', 'ci', '--omit=dev', '--ignore-scripts'], cwd=self.stage) (self.stage / '.release-commit').write_text(self.release + '\n') account = pwd.getpwnam(self.config['app_user']) self.run(['chown', '-R', f'{account.pw_uid}:{account.pw_gid}', str(self.stage)]) required = 2 * directory_size(self.app) + int(self.config['database_size_reserve_bytes']) if shutil.disk_usage(self.work).free < required + int(self.config['minimum_free_bytes']): raise ValueError('Dependency staging left insufficient backup/rollback space') except BaseException: self.log.close() raise stopped = False cutover = False try: # Stop only this application. Backups and migrations happen inside this maintenance window. stopped = True self.run([self.pm2, 'stop', self.config['pm2_app']]) self.backup_app(processes) cutover = True self.sync(self.stage) if migrate: target = next(row for row in processes if row['name'] == self.config['pm2_app']) database_env = {key: str(value) for key, value in target.get('pm2_env', {}).items() if key.startswith('DB_')} self.run(['runuser', '-u', self.config['app_user'], '--', 'env', 'PATH=' + self.env['PATH'], 'npm', 'run', 'migrate'], cwd=self.app, env={**self.env, **database_env}) self.run([self.pm2, 'restart', self.config['pm2_app']]) self.health() print(f'Deployment healthy: {self.release}', flush=True) except BaseException: if cutover: # Ensure the failed release cannot keep serving or writing during restoration. self.run([self.pm2, 'stop', self.config['pm2_app']]) self.restore_code() elif stopped: self.run([self.pm2, 'restart', self.config['pm2_app']]) self.health(rollback=True) raise finally: self.log.close() def main(payload): os.umask(0o077) config, action = payload['config'], payload['action'] app, work = validate_config(config) if action == 'inspect': inspect(config) return release = payload['release'] if not re.fullmatch(r'[0-9a-f]{40}', release): raise ValueError('Invalid release identifier') daemon_running(config) work.mkdir(parents=True, exist_ok=True, mode=0o700) with open(work / 'deploy.lock', 'a') as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) if action == 'stage': (work / 'incoming').mkdir(exist_ok=True, mode=0o700) (work / 'backups').mkdir(exist_ok=True, mode=0o700) return deployment = Deployment(config, release) if action == 'deploy': deployment.deploy(payload['checksum'], payload.get('migrate', False)) elif action == 'rollback': deployment.verify_target(require_online=False) deployment.verify_backup() deployment.run([deployment.pm2, 'stop', config['pm2_app']]) deployment.restore_code() else: raise ValueError('Unknown remote action') if __name__ == '__main__': def interrupted(signum, frame): raise InterruptedError('Deployment interrupted; recovery will be attempted if cutover started') signal.signal(signal.SIGTERM, interrupted) signal.signal(signal.SIGHUP, interrupted) try: main(json.loads(base64.b64decode(sys.argv[1]))) except (ValueError, KeyError, OSError, subprocess.CalledProcessError) as error: print(f'Remote deployment stopped: {error}', file=sys.stderr) sys.exit(1)