vh
80d982d1d8
The FV edge firewall was not backed up anywhere. Its config now lands in
/var/lib/restic/stage/fv-gateway-config.xml via ana-docker's pre-backup hook,
so the existing 01:00 restic snapshot captures it. ana-docker is one of the
three egress addresses the firewall's WAN allowlist permits, which is why the
pull lives there rather than with the FV hardware — a site that has lost power
cannot back itself up, and FV lost power two days ago.
Non-fatal by design: an unreachable firewall must not abort the nightly
database dumps. But a bad pull must not be promoted either. The summary loop
only rejects EMPTY staged files, and this endpoint answers an auth failure
with a perfectly non-empty HTML error page — which would have been backed up
as a firewall config that is the right size and restores nothing. The block
checks the body really contains <opnsense> and writes nothing otherwise.
Three tests cover it, including the HTML-error-page case. The first draft of
those tests was worthless: _fv returned a Path out of a TemporaryDirectory
context, so the tree was deleted before the assertions ran and every
exists()-is-False check passed regardless of what the script did. Only the
positive test failed, which is the sole reason the broken negatives were
caught. They now snapshot inside the tempdir's lifetime, and the docstring
says why.
Also records two OPNsense API lessons in docs/pfi/opnsense-api-reference.md:
endpoints are actions and must never be probed for existence by POSTing at
them — that is how /api/core/system/reboot took the FV site dark for 3.5
minutes while looking for an apply call this same file already documented —
and the apply step is service/reconfigure, which auth/user notably lacks, so
an API-only key edit persists in config.xml and does nothing until the OS user
sync runs at boot.
Credentials in /etc/restic/fv-gateway.env (root:600), template committed,
values vaulted as fv-gateway/opnsense-api-{key,secret}. Pre-change config
snapshot vaulted as fv-gateway/config-backup-20260914.
104 lines
6.1 KiB
Python
104 lines
6.1 KiB
Python
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
SCRIPT=Path(__file__).with_name('pre-backup.sh')
|
|
|
|
class BackupHookTests(unittest.TestCase):
|
|
def exercise(self, succeeds):
|
|
with tempfile.TemporaryDirectory(prefix='backup-hook-test-') as d:
|
|
root=Path(d); stage=root/'stage'; stage.mkdir(); binpath=root/'bin';binpath.mkdir()
|
|
previous=stage/'vaultwarden.pg_dump';previous.write_bytes(b'previous-good-backup')
|
|
docker=binpath/'docker';docker.write_text('#!/bin/sh\n[ "$1" = inspect ] && [ "$2" = vaultwarden ]\n');docker.chmod(0o755)
|
|
pg=binpath/'pg_dump';pg.write_text('#!/bin/sh\nprintf new-dump\nexit '+('0' if succeeds else '1')+'\n');pg.chmod(0o755)
|
|
env=dict(os.environ,PATH=str(binpath)+':'+os.environ['PATH'],RESTIC_STAGE_DIR=str(stage),RESTIC_DB_CREDS_FILE=str(root/'absent'),VW_PGPASS='fake',VW_PGHOST='fake',VW_PGUSER='fake',VW_PGDB='fake')
|
|
r=subprocess.run(['bash',str(SCRIPT)],env=env,capture_output=True,text=True)
|
|
if succeeds:
|
|
self.assertEqual(r.returncode,0,r.stdout+r.stderr)
|
|
self.assertEqual(previous.read_bytes(),b'new-dump')
|
|
else:
|
|
self.assertNotEqual(r.returncode,0,r.stdout+r.stderr)
|
|
self.assertEqual(previous.read_bytes(),b'previous-good-backup')
|
|
self.assertIn('required database dump(s) failed',r.stdout)
|
|
self.assertEqual(list(stage.glob('.pending.*')),[])
|
|
|
|
def test_failed_required_dump_preserves_previous_backup_and_cleans_scratch(self):
|
|
self.exercise(False)
|
|
|
|
def test_success_publishes_new_dump_and_cleans_scratch(self):
|
|
self.exercise(True)
|
|
|
|
def test_gitea_failure_removes_sql_scratch_and_exposes_error(self):
|
|
with tempfile.TemporaryDirectory(prefix='backup-gitea-test-') as d:
|
|
root=Path(d);stage=root/'stage';stage.mkdir();binpath=root/'bin';binpath.mkdir()
|
|
previous=stage/'gitea-dump.tar';previous.write_bytes(b'previous-good-archive')
|
|
docker=binpath/'docker'
|
|
docker.write_text('#!/bin/sh\nif [ "$1" = inspect ]; then [ "$2" = gitea ]; exit $?; fi\nshift 4\nexec "$@"\n')
|
|
docker.chmod(0o755)
|
|
gitea=binpath/'gitea'
|
|
gitea.write_text('#!/bin/sh\nwhile [ "$#" -gt 0 ]; do if [ "$1" = --tempdir ]; then shift; scratch=$1; fi; shift; done\nprintf %s "$scratch" > "$TEST_SCRATCH_PATH"\nprintf partial-sql > "$scratch/gitea-db.sql123"\necho simulated-export-failure >&2\nexit 9\n')
|
|
gitea.chmod(0o755)
|
|
path_record=root/'scratch-path'
|
|
env=dict(os.environ,PATH=str(binpath)+':'+os.environ['PATH'],RESTIC_STAGE_DIR=str(stage),RESTIC_DB_CREDS_FILE=str(root/'absent'),TEST_SCRATCH_PATH=str(path_record))
|
|
r=subprocess.run(['bash',str(SCRIPT)],env=env,capture_output=True,text=True)
|
|
self.assertNotEqual(r.returncode,0)
|
|
self.assertIn('simulated-export-failure',r.stderr)
|
|
self.assertFalse(Path(path_record.read_text()).exists())
|
|
self.assertEqual(previous.read_bytes(),b'previous-good-archive')
|
|
self.assertEqual(list(stage.glob('.pending.*')),[])
|
|
|
|
def _fv(self, curl_body, curl_rc=0):
|
|
"""Run the hook with a stubbed curl for the fv-gateway pull.
|
|
|
|
Returns a snapshot taken INSIDE the temp dir's lifetime. Returning the
|
|
Path instead is a trap: TemporaryDirectory deletes the tree on exit, so
|
|
every `exists()` assertion afterwards reads False and the negative tests
|
|
pass no matter what the script did. Caught 2026-09-14 — the positive
|
|
test failed, which is the only reason the broken negatives were noticed.
|
|
|
|
The guard under test is subtle: the summary loop only rejects EMPTY
|
|
staged files, and the OPNsense endpoint answers an auth failure with a
|
|
perfectly non-empty HTML error page. Without the content check that page
|
|
would be promoted and backed up as if it were a firewall config — a
|
|
backup that exists, is a plausible size, and restores nothing.
|
|
"""
|
|
with tempfile.TemporaryDirectory(prefix='backup-fv-test-') as d:
|
|
root=Path(d);stage=root/'stage';stage.mkdir();binpath=root/'bin';binpath.mkdir()
|
|
docker=binpath/'docker';docker.write_text('#!/bin/sh\nexit 1\n');docker.chmod(0o755)
|
|
curl=binpath/'curl'
|
|
curl.write_text('#!/bin/sh\nout=""\nwhile [ "$#" -gt 0 ]; do [ "$1" = -o ] && { shift; out=$1; }; shift; done\n'
|
|
'printf %s "$FV_STUB_BODY" > "$out"\nexit '+str(curl_rc)+'\n')
|
|
curl.chmod(0o755)
|
|
creds=root/'fv.env';creds.write_text('FV_API_KEY=k\nFV_API_SECRET=s\n')
|
|
env=dict(os.environ,PATH=str(binpath)+':'+os.environ['PATH'],
|
|
RESTIC_STAGE_DIR=str(stage),RESTIC_DB_CREDS_FILE=str(root/'absent'),
|
|
RESTIC_FV_CREDS_FILE=str(creds),FV_STUB_BODY=curl_body)
|
|
r=subprocess.run(['bash',str(SCRIPT)],env=env,capture_output=True,text=True)
|
|
cfg=stage/'fv-gateway-config.xml'
|
|
return {'rc':r.returncode,'out':r.stdout,'err':r.stderr,
|
|
'staged':sorted(p.name for p in stage.iterdir()),
|
|
'body':cfg.read_text() if cfg.exists() else None}
|
|
|
|
def test_fv_gateway_config_is_staged_when_body_is_a_real_config(self):
|
|
g=self._fv('<?xml version="1.0"?><opnsense><system/></opnsense>')
|
|
self.assertEqual(g['rc'],0,g['out']+g['err'])
|
|
self.assertIn('fv-gateway-config.xml',g['staged'])
|
|
self.assertIn('<opnsense>',g['body'])
|
|
|
|
def test_fv_gateway_html_error_page_is_never_staged(self):
|
|
g=self._fv('<!DOCTYPE html><html><body>401 Unauthorized</body></html>')
|
|
self.assertEqual(g['rc'],0,g['out']+g['err']) # non-fatal by design
|
|
self.assertEqual(g['staged'],[])
|
|
self.assertIsNone(g['body'])
|
|
self.assertIn('not an OPNsense config',g['err'])
|
|
|
|
def test_fv_gateway_unreachable_is_non_fatal_and_stages_nothing(self):
|
|
g=self._fv('',curl_rc=7)
|
|
self.assertEqual(g['rc'],0,g['out']+g['err'])
|
|
self.assertEqual(g['staged'],[])
|
|
self.assertIn('config pull failed',g['err'])
|
|
|
|
if __name__=='__main__':unittest.main()
|