From 571523354ee45f5ad140bb154eb20a29d3d2e506 Mon Sep 17 00:00:00 2001
From: Jonas Braathen <jonas@ponas.no>
Date: Mon, 2 Nov 2026 01:47:37 +0100
Subject: [PATCH] fix(zone_sync): replace zone files in one step

---
 mdb/services/zone_sync.py | 34 ++++++++++++++++++++++++++++++++--
 tests/test_zone_sync.py   | 34 +++++++++++++++++++++++++++++++++-
 2 files changed, 65 insertions(+), 3 deletions(-)

diff --git a/mdb/services/zone_sync.py b/mdb/services/zone_sync.py
index 7532d37..d5be59e 100644
--- a/mdb/services/zone_sync.py
+++ b/mdb/services/zone_sync.py
@@ -1,6 +1,7 @@
 import os
 import re
 import shutil
+import stat
 import sys
 import tempfile
 from difflib import context_diff
@@ -124,8 +125,7 @@ class ZoneSyncService:
             path = "/".join(zonefile.split("/")[:-1])
             os.path.isdir(path) or os.makedirs(path)
 
-        with open(zonefile, "w") as f:
-            f.write(new)
+        self.write_zone_file(zonefile, new)
 
         if not self.debug:
             zone.domain_active_serial = zone.domain_serial
@@ -135,6 +135,36 @@ class ZoneSyncService:
         self.changes[zone.domain_name]["success"] = True
         return True
 
+    def write_zone_file(self, path, contents):
+        """Replace the file at path with contents, whole or not at all.
+
+        named may read the file at any time, and a write in place could leave it half
+        written. The new file takes the old one's owner and mode: bind owns the zone
+        files, and this runs as root.
+        """
+        directory, name = os.path.split(path)
+        fd, tmp = tempfile.mkstemp(dir=directory or ".", prefix=f".{name}.")
+        try:
+            with os.fdopen(fd, "w") as f:
+                f.write(contents)
+                f.flush()
+                os.fsync(f.fileno())
+            try:
+                st = os.stat(path)
+            except FileNotFoundError:
+                os.chmod(tmp, 0o644)
+            else:
+                os.chmod(tmp, stat.S_IMODE(st.st_mode))
+                if (st.st_uid, st.st_gid) != (os.getuid(), os.getgid()):
+                    os.chown(tmp, st.st_uid, st.st_gid)
+            os.replace(tmp, path)
+        except BaseException:
+            try:
+                os.unlink(tmp)
+            except FileNotFoundError:
+                pass
+            raise
+
     def read_serial(self, contents):
         """Return the SOA serial of a zone file, or None if there is none."""
         match = re.search(r"^\s*(\d+)\s*;\s*serial", contents, re.MULTILINE)
diff --git a/tests/test_zone_sync.py b/tests/test_zone_sync.py
index e795b5a..732c1cd 100644
--- a/tests/test_zone_sync.py
+++ b/tests/test_zone_sync.py
@@ -1,6 +1,6 @@
 import os
 import subprocess
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 import pytest
 from freezegun import freeze_time
@@ -339,3 +339,35 @@ def test_sync_zone_does_not_write_a_zone_that_fails_validation(db_domain, tmp_pa
     change = svc.changes["example.org"]
     assert change["success"] is False
     assert "not-an-address" in change["errors"]
+
+
+def test_write_zone_file_replaces_file_and_keeps_its_mode(service, tmp_path):
+    path = tmp_path / "db.example.org"
+    path.write_text("old\n")
+    path.chmod(0o664)
+    inode = path.stat().st_ino
+
+    service.write_zone_file(str(path), "new\n")
+
+    assert path.read_text() == "new\n"
+    assert path.stat().st_mode & 0o777 == 0o664
+    assert path.stat().st_ino != inode  # renamed over, not written in place
+    assert os.listdir(tmp_path) == ["db.example.org"]
+
+
+def test_write_zone_file_creates_missing_file(service, tmp_path):
+    path = tmp_path / "db.example.org"
+    service.write_zone_file(str(path), "new\n")
+    assert path.read_text() == "new\n"
+    assert path.stat().st_mode & 0o777 == 0o644
+
+
+def test_write_zone_file_leaves_old_file_when_write_fails(service, tmp_path):
+    path = tmp_path / "db.example.org"
+    path.write_text("old\n")
+
+    with patch("os.fsync", side_effect=OSError("disk full")), pytest.raises(OSError):
+        service.write_zone_file(str(path), "new\n")
+
+    assert path.read_text() == "old\n"
+    assert os.listdir(tmp_path) == ["db.example.org"]