"""V8.18 recovery safety tests; never touch a real database/server.""" import hashlib import io import json import subprocess import zipfile import pytest from pathlib import Path from app import backup, db, jobs, enterprise, pg_disaster_recovery as pg URL='postgresql://backup_user:super-secret@127.0.0.1:5432/empty_restore?sslmode=require' def test_pg_env_never_passes_secret_in_commands(): env=pg.connection_env(URL, {'PATH':'/usr/bin'}) assert env['PGPASSWORD']=='super-secret' assert env['PGHOST']=='127.0.0.1' assert env['PGSSLMODE']=='require' assert env['PGPASSFILE']=='/dev/null' with pytest.raises(pg.RecoveryError):pg.connection_env('not a pg url') with pytest.raises(pg.RecoveryError):pg.connection_env(URL+'&danger=1') def test_sqlite_backup_cannot_run_in_postgres_mode(monkeypatch,tmp_path): monkeypatch.setattr(db,'POSTGRES',True) monkeypatch.setattr(db,'DATA',tmp_path) monkeypatch.setattr(db,'DB_FILE',tmp_path/'no-fake.sqlite3') with pytest.raises(RuntimeError,match='PostgreSQL'): backup.backup() assert not (tmp_path/'no-fake.sqlite3').exists() def _fake_tools(commands): def fake_run(args,env,timeout=1800): assert 'super-secret' not in ' '.join(args) commands.append(args) if args[0]=='pg_dump': Path(args[-1]).write_bytes(b'PGDMP\0offline-mock') return subprocess.CompletedProcess(args,0,b'',b'') if args[0]=='psql': return subprocess.CompletedProcess(args,0,b'0\n',b'') return subprocess.CompletedProcess(args,0,b'',b'') return fake_run def test_backup_verify_and_restore_to_empty_destination(tmp_path,monkeypatch): storage=tmp_path/'storage';(storage/'media').mkdir(parents=True) (storage/'media'/'logo.png').write_bytes(b'fake-png') archive=tmp_path/'external'/'backup.zip';commands=[] monkeypatch.setattr(pg,'run_pg',_fake_tools(commands)) with pytest.raises(pg.RecoveryError,match='Stop all'): pg.backup(URL,storage,archive) result=pg.backup(URL,storage,archive,confirmed_offline=True) assert result['files']==2 manifest=pg.verify(archive) assert manifest['database']=='postgresql' assert 'media/logo.png' in manifest['files'] assert [x[0] for x in commands[:2]]==['pg_dump','pg_restore'] with pytest.raises(pg.RecoveryError,match='overwrite'): pg.backup(URL,storage,archive,confirmed_offline=True) target=tmp_path/'restored' with pytest.raises(pg.RecoveryError,match='confirmation'): pg.restore_new(URL,archive,target) restored=pg.restore_new(URL,archive,target,confirmed_offline=True) assert restored['media_files']==1 assert (target/'media'/'logo.png').read_bytes()==b'fake-png' assert not (target/'database.dump').exists() assert [x[0] for x in commands[-2:]]==['psql','pg_restore'] with pytest.raises(pg.RecoveryError,match='must not exist'): pg.restore_new(URL,archive,target,confirmed_offline=True) def test_tamper_rejected_before_db_restore(tmp_path,monkeypatch): src=tmp_path/'media';src.mkdir() z=tmp_path/'good.zip' monkeypatch.setattr(pg,'run_pg',_fake_tools([])) pg.backup(URL,tmp_path,z,confirmed_offline=True) corrupt=tmp_path/'corrupt.zip' with zipfile.ZipFile(z) as original, zipfile.ZipFile(corrupt,'w') as out: for name in original.namelist(): payload=original.read(name) if name=='database.dump':payload=b'PODMP\0offline-mock' out.writestr(name,payload) with pytest.raises(pg.RecoveryError,match='Checksum'): pg.verify(corrupt) assert not (tmp_path/'restored').exists() def test_traversal_and_symlink_refused(tmp_path,monkeypatch): src=tmp_path/'storage';src.mkdir() (src/'media').symlink_to(tmp_path) monkeypatch.setattr(pg,'run_pg',_fake_tools([])) with pytest.raises(pg.RecoveryError,match='Symlink'): pg.backup(URL,src,tmp_path/'bad.zip',confirmed_offline=True) evil=tmp_path/'evil.zip' with zipfile.ZipFile(evil,'w') as z: z.writestr('manifest.json','{}') z.writestr('database.dump',b'PGDMP') z.writestr('../hijack',b'evil') with pytest.raises(pg.RecoveryError,match='Unsafe'): pg.verify(evil) def test_expired_release_is_never_auto_retried(tmp_path,monkeypatch): monkeypatch.setattr(db,'POSTGRES',False) monkeypatch.setattr(db,'DATA',tmp_path) monkeypatch.setattr(db,'DB_FILE',tmp_path/'siteforge.sqlite3') db.init(); enterprise.setup() db.insert('INSERT INTO sites(id,name,created_at,updated_at) VALUES (?,?,?,?)', ('site818','Sample',db.now(),db.now())) for jobid,kind in [('release818','remote_deploy'),('roll818','remote_rollback'),('build818','build')]: db.insert('''INSERT INTO build_jobs(id,site_id,kind,state,created_by,created_at, attempts,max_attempts,worker_id,lease_until) VALUES (?,?,?,?,?,?,?,?,?,?)''', (jobid,'site818',kind,'running','system',db.now(),1,3,'dead-worker','2020-01-01T00:00:00+00:00')) assert jobs.recover_expired()==3 assert db.one('SELECT state FROM build_jobs WHERE id=?',('release818',))['state']=='failed' assert db.one('SELECT state FROM build_jobs WHERE id=?',('roll818',))['state']=='failed' assert db.one('SELECT state FROM build_jobs WHERE id=?',('build818',))['state']=='retrying' assert 'reconcile' in db.one('SELECT error FROM build_jobs WHERE id=?',('release818',))['error'] def test_unknown_remote_outcome_is_never_retried(tmp_path,monkeypatch): from app import worker monkeypatch.setattr(db,'POSTGRES',False) monkeypatch.setattr(db,'DATA',tmp_path) monkeypatch.setattr(db,'DB_FILE',tmp_path/'siteforge.sqlite3') db.init();enterprise.setup() db.insert('INSERT INTO sites(id,name,created_at,updated_at) VALUES (?,?,?,?)', ('release-site','Sample',db.now(),db.now())) for jid in ('release-retry','release-cancel'): db.insert('''INSERT INTO build_jobs(id,site_id,kind,state,created_by,created_at, attempts,max_attempts,worker_id,lease_until) VALUES (?,?,?,?,?,?,?,?,?,?)''', (jid,'release-site','remote_deploy','running','system',db.now(),1,3, worker.WORKER_ID,'2030-01-01T00:00:00+00:00')) old=db.one('SELECT * FROM build_jobs WHERE id=?',('release-retry',)) worker.complete(old,error='network error',retryable=True) assert db.one('SELECT state FROM build_jobs WHERE id=?',('release-retry',))['state']=='failed' old=db.one('SELECT * FROM build_jobs WHERE id=?',('release-cancel',)) worker.complete(old,cancelled=True) state=db.one('SELECT state,error FROM build_jobs WHERE id=?',('release-cancel',)) assert state['state']=='failed' and 'reconcile' in state['error']