fix(cve-triage): unique tmp filename for wafgen write_category

Two concurrent regenerators (CLI cron + panel) writing waf-rules.json
both used a single fixed .json.tmp name, letting one process's
os.replace() promote the other's in-flight content — a silent lost
update. Use tempfile.mkstemp() in the same directory to get a unique
temp path per call, and clean it up on any failure before the swap.

Co-Authored-By: Gerald KERMA <devel@cybermind.fr>
This commit is contained in:
CyberMind-FR 2026-07-18 07:35:22 +02:00
parent eb727669e2
commit 8becec8480
2 changed files with 72 additions and 3 deletions

View File

@ -17,6 +17,7 @@ from __future__ import annotations
import json
import os
import re
import tempfile
from pathlib import Path
from .extract import Candidate
@ -57,6 +58,19 @@ def write_category(rules_path: Path, candidates: list[Candidate], *, now: str) -
cats = data.setdefault("categories", {})
cats[CATEGORY_KEY] = build_category(candidates, now=now)
tmp = rules_path.with_suffix(".json.tmp")
tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
os.replace(tmp, rules_path)
# Unique temp file in the SAME directory as rules_path: concurrent writers
# (CLI cron regen + panel-triggered regen) must never collide on a single
# fixed name, and os.replace() only stays atomic on the same filesystem.
fd, tmp_name = tempfile.mkstemp(
dir=rules_path.parent, prefix=".waf-rules-", suffix=".tmp"
)
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
f.write(json.dumps(data, indent=2, ensure_ascii=False) + "\n")
os.replace(tmp_name, rules_path)
except BaseException:
try:
os.unlink(tmp_name)
except FileNotFoundError:
pass
raise

View File

@ -3,6 +3,9 @@
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
import json
import os
import pytest
from api.wafgen.emit import CATEGORY_KEY, build_category, write_category
from api.wafgen.extract import Candidate
@ -76,3 +79,55 @@ def test_empty_candidates_writes_an_empty_category(tmp_path):
write_category(p, [], now="2026-07-18T00:00:00Z")
d = json.loads(p.read_text())
assert d["categories"][CATEGORY_KEY]["patterns"] == []
def test_temp_file_path_is_unique_per_call(tmp_path, monkeypatch):
# Concurrent writers (CLI cron regen + panel regen) must not share a
# single fixed ".json.tmp" name: two writers racing on the same fixed
# name means one process's os.replace() can promote the OTHER
# process's in-flight content (a silent lost update). Spy on the
# actual src path passed to os.replace() across two calls — with a
# fixed name this would be identical every time; with mkstemp() it
# must differ.
import api.wafgen.emit as emit_mod
p = _rules_file(tmp_path)
real_replace = os.replace
seen_src_paths = []
def spy_replace(src, dst):
seen_src_paths.append(str(src))
return real_replace(src, dst)
monkeypatch.setattr(emit_mod.os, "replace", spy_replace)
write_category(p, [cand()], now="2026-07-18T00:00:00Z")
write_category(p, [cand()], now="2026-07-18T00:00:00Z")
assert len(seen_src_paths) == 2
assert seen_src_paths[0] != seen_src_paths[1]
def test_no_leftover_tmp_file_after_successful_write(tmp_path):
p = _rules_file(tmp_path)
write_category(p, [cand()], now="2026-07-18T00:00:00Z")
leftovers = [f.name for f in tmp_path.iterdir() if f.name != p.name]
assert leftovers == []
def test_write_category_raises_on_missing_rules_file(tmp_path):
missing = tmp_path / "waf-rules.json"
with pytest.raises(FileNotFoundError):
write_category(missing, [cand()], now="2026-07-18T00:00:00Z")
def test_write_category_raises_on_corrupt_rules_file_and_leaves_it_unchanged(tmp_path):
p = tmp_path / "waf-rules.json"
corrupt = "{ not valid json"
p.write_text(corrupt)
with pytest.raises(json.JSONDecodeError):
write_category(p, [cand()], now="2026-07-18T00:00:00Z")
# A generator must never blank the WAF's live rules on failure.
assert p.read_text() == corrupt
leftovers = [f.name for f in tmp_path.iterdir() if f.name != p.name]
assert leftovers == []