fix(db): WAL-safe safety backup + tests for validate-first and zip-traversal guard

This commit is contained in:
2026-07-02 06:45:17 +00:00
parent 7edd0a1265
commit 280b3d25ec
2 changed files with 31 additions and 1 deletions
+1 -1
View File
@@ -78,7 +78,7 @@ def restore_db_and_waveforms(db_path: Path, waveform_root: Path,
if db_path.exists(): if db_path.exists():
ts = datetime.datetime.utcnow().strftime("%Y%m%d_%H%M%S") ts = datetime.datetime.utcnow().strftime("%Y%m%d_%H%M%S")
safety = db_path.with_name(f"{db_path.name}.pre-restore-{ts}") safety = db_path.with_name(f"{db_path.name}.pre-restore-{ts}")
shutil.copy2(db_path, safety) snapshot_db(db_path, safety)
safety_name = safety.name safety_name = safety.name
# Clear the OUTGOING db's WAL sidecars so they cannot attach to the incoming file. # Clear the OUTGOING db's WAL sidecars so they cannot attach to the incoming file.
+30
View File
@@ -129,5 +129,35 @@ def test_restore_rejects_non_sqlite(tmp_path):
bogus.write_bytes(b"not a sqlite file at all") bogus.write_bytes(b"not a sqlite file at all")
wf_root = tmp_path / "waveforms" wf_root = tmp_path / "waveforms"
wf_root.mkdir() wf_root.mkdir()
original_bytes = db_path.read_bytes()
with pytest.raises(ValueError): with pytest.raises(ValueError):
restore_db_and_waveforms(db_path, wf_root, bogus, None) restore_db_and_waveforms(db_path, wf_root, bogus, None)
assert db_path.read_bytes() == original_bytes
assert list(tmp_path.glob("*.pre-restore-*")) == []
def test_restore_skips_unsafe_zip_members(tmp_path):
import zipfile
from sfm.db_snapshot import restore_db_and_waveforms
db_path = tmp_path / "seismo_relay.db"
_make_db(db_path, rows=1)
new_db = tmp_path / "incoming.db"
_make_db(new_db, rows=2)
wf_root = tmp_path / "waveforms"
wf_root.mkdir()
wf_zip = tmp_path / "wf.zip"
with zipfile.ZipFile(wf_zip, "w") as zf:
zf.writestr("BE111/ok.h5", b"good")
zf.writestr("../evil.txt", b"escape via parent dir")
zf.writestr("/tmp/abs_evil.txt", b"escape via absolute path")
zf.writestr("baddir/", b"")
result = restore_db_and_waveforms(db_path, wf_root, new_db, wf_zip)
assert (wf_root / "BE111" / "ok.h5").exists()
assert result["waveforms_unpacked"] == 1
assert not (wf_root.parent / "evil.txt").exists()
assert not Path("/tmp/abs_evil.txt").exists()