diff --git a/MethodenAnalyser3.py b/MethodenAnalyser3.py
index c06dd5a..847be23 100644
--- a/MethodenAnalyser3.py
+++ b/MethodenAnalyser3.py
@@ -11,7 +11,9 @@
import pkgutil
import re
import sqlite3
+import stat
import sys
+import tempfile
import threading
import tkinter as tk
import warnings
@@ -1490,10 +1492,9 @@ def _remove_unused_imports(source: str, tree: ast.AST, unused_set: set[str]) ->
"""Remove AST import spans without deleting neighboring executable code."""
raw = source.encode("utf-8")
offsets = [0]
- # AST columns count UTF-8 bytes; only LF advances AST line numbers.
- # str.splitlines() would incorrectly count form-feed as a new line.
- for line in raw.split(b"\n")[:-1]:
- offsets.append(offsets[-1] + len(line) + 1)
+ # AST columns count UTF-8 bytes. Recognize physical CR/LF boundaries
+ # without counting form-feed or rewriting untouched newline sequences.
+ offsets.extend(match.end() for match in re.finditer(rb"\r\n|\r|\n", raw))
nodes = sorted(_unused_import_nodes(tree, unused_set),
key=lambda node: (node.lineno, node.col_offset), reverse=True)
changed_lines: set[int] = set()
@@ -1519,6 +1520,64 @@ def _remove_unused_imports(source: str, tree: ast.AST, unused_set: set[str]) ->
return updated, changed_lines
+def _stage_auto_fix_bytes(path: pathlib.Path, data: bytes) -> pathlib.Path:
+ """Fully write and flush a same-directory temporary file."""
+ staged = None
+ complete = False
+ try:
+ with tempfile.NamedTemporaryFile(mode="wb", dir=path.parent,
+ prefix=f".{path.name}.autofix-", delete=False) as handle:
+ staged = pathlib.Path(handle.name)
+ if handle.write(data) != len(data):
+ raise OSError("Incomplete Auto-Fix write")
+ handle.flush()
+ os.fsync(handle.fileno())
+ complete = True
+ return staged
+ finally:
+ if staged is not None and not complete:
+ staged.unlink(missing_ok=True)
+
+
+def _commit_auto_fix(path: str, original: bytes, updated: bytes) -> str:
+ """Publish a complete backup exclusively, then atomically replace the source."""
+ source = pathlib.Path(path).resolve(strict=True)
+ mode = stat.S_IMODE(source.stat().st_mode)
+ staged_source = staged_backup = None
+ try:
+ staged_source = _stage_auto_fix_bytes(source, updated)
+ staged_backup = _stage_auto_fix_bytes(source, original)
+ if source.read_bytes() != original:
+ raise OSError("Source changed during Auto-Fix")
+ index = 0
+ while True:
+ suffix = f".bak.{index}" if index else ".bak"
+ backup = pathlib.Path(str(source) + suffix)
+ try:
+ # Same-directory hard-link publication is atomic and exclusive:
+ # an existing or concurrently created recovery copy is preserved.
+ os.link(staged_backup, backup)
+ except FileExistsError:
+ index += 1
+ else:
+ break
+ os.chmod(staged_source, mode)
+ if source.read_bytes() != original:
+ raise OSError("Source changed during Auto-Fix")
+ os.replace(staged_source, source)
+ return str(backup)
+ finally:
+ for staged in (staged_source, staged_backup):
+ if staged is not None:
+ try:
+ # A staged source may carry the original read-only flag.
+ # Clear it only on our temporary path before cleanup.
+ staged.chmod(stat.S_IRUSR | stat.S_IWUSR)
+ except FileNotFoundError:
+ continue
+ staged.unlink(missing_ok=True)
+
+
def auto_fix_unused_imports(output_widget: scrolledtext.ScrolledText, status_widget: tk.Label | None = None) -> None:
"""
Entfernt ungenutzte Imports aus der zuletzt analysierten Datei.
@@ -1551,24 +1610,22 @@ def auto_fix_unused_imports(output_widget: scrolledtext.ScrolledText, status_wid
try:
# Datei lesen mit Encoding-Fallback — erkanntes Encoding für Schreibzugriff merken
detected_encoding = "utf-8"
+ original_bytes = pathlib.Path(_last_analysis_path).read_bytes()
try:
- with open(_last_analysis_path, "r", encoding="utf-8") as f:
- lines = f.readlines()
+ original_text = original_bytes.decode("utf-8")
except UnicodeDecodeError:
detected_encoding = "latin-1"
- with open(_last_analysis_path, "r", encoding="latin-1") as f:
- lines = f.readlines()
-
- # AST parsen (readlines() beibehalten — splitlines() würde bei \x0c
- # Zeilennummern gegenüber AST-lineno verschieben und falsche Zeilen löschen)
- tree = ast.parse("".join(lines))
+ original_text = original_bytes.decode("latin-1")
+ # Parse the original text; physical byte offsets preserve mixed
+ # newline sequences and avoid treating form-feed as a new line.
+ tree = ast.parse(original_text)
# Import-Zeilen markieren die entfernt werden sollen
# Findings may be stale after an editor change or the confirmation
# dialog. Remove only names approved earlier AND still unused now.
- current_result = analyze_source("".join(lines), _last_analysis_path)
+ current_result = analyze_source(original_text, _last_analysis_path)
unused_set = set(_last_analysis_result.unused_imports) & set(current_result.unused_imports)
- updated_source, lines_to_remove = _remove_unused_imports("".join(lines), tree, unused_set)
+ updated_source, lines_to_remove = _remove_unused_imports(original_text, tree, unused_set)
if not lines_to_remove:
if status_widget is not None:
@@ -1581,13 +1638,8 @@ def auto_fix_unused_imports(output_widget: scrolledtext.ScrolledText, status_wid
# Backup und Ausgabe im erkannten Encoding — verhindert Korrumpierung von
# latin-1-Dateien mit nicht-ASCII-Zeichen und # coding: latin-1 Deklaration
- backup_path = _last_analysis_path + ".bak"
- with open(backup_path, "w", encoding=detected_encoding) as f:
- f.writelines(lines)
-
- # Neue Datei ohne ungenutzte Imports
- with open(_last_analysis_path, "w", encoding=detected_encoding) as f:
- f.write(updated_source)
+ updated_bytes = updated_source.encode(detected_encoding)
+ backup_path = _commit_auto_fix(_last_analysis_path, original_bytes, updated_bytes)
# Ausgabe
output_widget.insert(tk.END, f"\n\n{_t('gui_autofix_success_header')}\n")
diff --git a/README.md b/README.md
index 5d0a298..96db3f3 100644
--- a/README.md
+++ b/README.md
@@ -5,7 +5,7 @@
- 
+ 
@@ -173,7 +173,7 @@ MethodenAnalyser enforces 10 strict architectural, runtime, and security invaria
| **INV-ENCODING-04** | Lossless Multi-Encoding Resilience | Default UTF-8 reading with automatic Latin-1 fallback. Never corrupts original character encodings, UTF-8 BOMs, or German umlauts. |
| **INV-ISOLATION-05** | Loopback Web Companion Isolation | Local helper companion binds strictly to `127.0.0.1` by default. Deliberate `--host 0.0.0.0` LAN test mode requires explicit user configuration. |
| **INV-TRAVERSAL-06** | Anti-Path-Traversal & ZIP Safety | Web companion rejects path traversal (`..`), bounds member counts, and enforces maximum archive byte limits to prevent ZIP bombs. |
-| **INV-AUTOFIX-07** | Reversible Auto-Fix with Backup | Auto-Fix requires explicit confirmation, rechecks current source usage and creates a `.bak` backup before writing modifications. |
+| **INV-AUTOFIX-07** | Reversible Auto-Fix with Backup | Auto-Fix requires explicit confirmation, rechecks current source usage and publishes a byte-exact backup without replacing existing copies (`.bak`, `.bak.1`, …), then atomically replaces the source from a flushed temporary file. Detected source changes abort the write; encoding, newline style and file mode are retained. |
| **INV-PLATFORM-08** | Multi-Platform Parity | 100% standard library core verified across Windows Server 2025, Ubuntu Linux, and macOS 26. |
| **INV-SYNC-09** | Cloud-Sync & Multi-Agent Lock Safety | Fails closed on lock detection (`LOCK.*`), ignores cloud sync conflict files (`*-conflict-*`), preventing concurrency races. |
| **INV-SLA-10** | 48h Security SLA & 5-Day Triage | Strict vulnerability commitment: 48-hour initial response SLA and 5-business-day triage commitment with coordinated disclosure. |
diff --git a/README_de.md b/README_de.md
index 366c144..c2b96c0 100644
--- a/README_de.md
+++ b/README_de.md
@@ -5,7 +5,7 @@
- 
+ 
@@ -173,7 +173,7 @@ MethodenAnalyser garantiert 10 verbindliche Architektur-, Laufzeit- und Sicherhe
| **INV-ENCODING-04** | Verlustfreie Multi-Encoding-Resilienz | Liest Dateien standardmäßig als UTF-8 mit automatischem Latin-1 Fallback. Verändert niemals Original-Encodings oder deutsche Umlaute. |
| **INV-ISOLATION-05** | Loopback Web-Begleiter-Isolation | Der lokale Hilfsserver bindet sich standardmäßig ausschließlich an `127.0.0.1`. Der LAN-Modus `--host 0.0.0.0` erfordert explizite Aktivierung. |
| **INV-TRAVERSAL-06** | Pfad-Traversal- & ZIP-Sicherheit | Der Web-Begleiter blockiert Directory-Traversal (`..`), begrenzt Dateizahlen und unkomprimierte Byte-Mengen zur Abwehr von ZIP-Bomben. |
-| **INV-AUTOFIX-07** | Reversibler Auto-Fix mit Backup | Auto-Fix erfordert eine explizite Bestätigung, prüft die aktuelle Verwendung erneut und legt vor jeder Änderung eine `.bak`-Sicherheitskopie an. |
+| **INV-AUTOFIX-07** | Reversibler Auto-Fix mit Backup | Auto-Fix erfordert eine explizite Bestätigung, prüft die aktuelle Verwendung erneut und veröffentlicht eine bytegenaue Sicherung ohne bestehende Backups zu überschreiben (`.bak`, `.bak.1`, …) und ersetzt die Quelle anschließend atomar aus einer vollständig geschriebenen temporären Datei. Erkannte Quelländerungen brechen den Schreibvorgang ab; Kodierung, Zeilenumbruchstil und Dateimodus bleiben erhalten. |
| **INV-PLATFORM-08** | Plattformübergreifende Parität | Vollständige Standardbibliothek-Kompatibilität, verifiziert auf Windows Server 2025, Ubuntu Linux und macOS 26. |
| **INV-SYNC-09** | Cloud-Sync- & Lock-Resilienz | Beachtet Dateisperren (`LOCK.*`) im Fail-Closed-Verfahren und ignoriert Cloud-Sync-Konfliktdateien (`*-conflict-*`). |
| **INV-SLA-10** | 48h Sicherheits-SLA & 5-Tage-Triage | Verbindliches Sicherheitsversprechen: Eingangsbestätigung binnen 48 Stunden, Triage und Behebungsplan binnen 5 Werktagen. |
diff --git a/llms.txt b/llms.txt
index d9048e9..ea5b237 100644
--- a/llms.txt
+++ b/llms.txt
@@ -35,7 +35,7 @@ The tool strictly adheres to 10 architectural and security invariants:
- INV-ENCODING-04: Lossless Multi-Encoding Resilience (UTF-8 default with Latin-1 fallback)
- INV-ISOLATION-05: Loopback Web Companion Isolation (binds to 127.0.0.1 by default)
- INV-TRAVERSAL-06: Path Traversal & ZIP Safety Bounds (blocks `..` and bounds member sizes)
-- INV-AUTOFIX-07: Reversible Auto-Fix with `.bak` Backup; current source is rechecked after confirmation, and only approved imports that remain unused are removed. Import spans preserve neighboring statements; empty suites retain pass, and the result is syntax-checked before writing.
+- INV-AUTOFIX-07: Reversible Auto-Fix with `.bak` Backup; current source is rechecked after confirmation, and only approved imports that remain unused are removed. Import spans preserve neighboring statements; empty suites retain pass, and the result is syntax-checked before writing. Auto-Fix stages and flushes complete byte payloads, exclusively publishes numbered backups via same-directory hard links and atomically replaces the source. Detected source changes abort; original bytes remain in the backup. Unsupported backup publication fails before source replacement.
- INV-PLATFORM-08: Cross-Platform Parity across Windows, Linux, and macOS
- INV-SYNC-09: Cloud-Sync & Multi-Agent Lock Resilience (fail-closed against `LOCK.*`)
- INV-SLA-10: 48h Initial Response & 5-Business-Day Triage SLA
diff --git a/tests/test_autofix_atomic_write.py b/tests/test_autofix_atomic_write.py
new file mode 100644
index 0000000..dff83f3
--- /dev/null
+++ b/tests/test_autofix_atomic_write.py
@@ -0,0 +1,157 @@
+"""Auto-fix must preserve source bytes and existing recovery copies on failure."""
+import stat
+from contextlib import contextmanager
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+import MethodenAnalyser3 as m
+
+
+def prepare(monkeypatch, path, raw):
+ path.write_bytes(raw)
+ monkeypatch.setattr(m, "_last_analysis_path", str(path))
+ monkeypatch.setattr(m, "_last_analysis_result", m.analyze_file(str(path)))
+ dialogs = Mock()
+ dialogs.askyesno.return_value = True
+ monkeypatch.setattr(m, "messagebox", dialogs)
+ return dialogs
+
+
+def test_existing_backup_is_not_overwritten(monkeypatch, tmp_path):
+ path = tmp_path / "source.py"
+ raw = b"import os\nanswer = 42\n"
+ dialogs = prepare(monkeypatch, path, raw)
+ backup = path.with_suffix(".py.bak")
+ backup.write_bytes(b"older recovery copy")
+ m.auto_fix_unused_imports(Mock())
+ dialogs.showerror.assert_not_called()
+ assert path.read_text(encoding="utf-8") == "answer = 42\n"
+ assert backup.read_bytes() == b"older recovery copy"
+ assert path.with_suffix(".py.bak.1").read_bytes() == raw
+
+
+@pytest.mark.parametrize("mode", [0o644, 0o444])
+def test_publication_failure_preserves_original(monkeypatch, tmp_path, mode):
+ path = tmp_path / "source.py"
+ raw = b"import os\nanswer = 42\n"
+ dialogs = prepare(monkeypatch, path, raw)
+ path.chmod(mode)
+ original_mode = stat.S_IMODE(path.stat().st_mode)
+ def fail(*args):
+ raise OSError("publication failed")
+ monkeypatch.setattr(m.os, "replace", fail)
+ m.auto_fix_unused_imports(Mock())
+ assert path.read_bytes() == raw
+ assert stat.S_IMODE(path.stat().st_mode) == original_mode
+ assert dialogs.showerror.call_count == 1
+ dialogs.showinfo.assert_not_called()
+ assert not list(tmp_path.glob(".*.autofix-*"))
+ path.chmod(0o644)
+
+
+@pytest.mark.parametrize("newline", [b"\r\n", b"\n", b"\r"])
+def test_backup_preserves_exact_newline_bytes(monkeypatch, tmp_path, newline):
+ path = tmp_path / "source.py"
+ raw = b"import os" + newline + b"answer = 42" + newline
+ dialogs = prepare(monkeypatch, path, raw)
+ m.auto_fix_unused_imports(Mock())
+ dialogs.showerror.assert_not_called()
+ assert path.with_suffix(".py.bak").read_bytes() == raw
+ assert path.read_bytes() == b"answer = 42" + newline
+
+
+@pytest.mark.parametrize("failure", ["partial", "fsync", "backup_link"])
+def test_staging_failure_preserves_source_and_existing_backup(monkeypatch, tmp_path, failure):
+ path = tmp_path / "source.py"
+ raw = b"import os\nanswer = 42\n"
+ dialogs = prepare(monkeypatch, path, raw)
+ backup = path.with_suffix(".py.bak")
+ backup.write_bytes(b"older backup")
+ def fail(*args):
+ raise OSError("injected failure")
+ if failure == "partial":
+ real_temp = m.tempfile.NamedTemporaryFile
+ @contextmanager
+ def partial_temp(*args, **kwargs):
+ with real_temp(*args, **kwargs) as handle:
+ yield SimpleNamespace(name=handle.name, flush=handle.flush, fileno=handle.fileno,
+ write=lambda data: handle.write(data[:len(data) // 2]))
+ monkeypatch.setattr(m.tempfile, "NamedTemporaryFile", partial_temp)
+ elif failure == "fsync":
+ monkeypatch.setattr(m.os, "fsync", fail)
+ else:
+ monkeypatch.setattr(m.os, "link", fail)
+ m.auto_fix_unused_imports(Mock())
+ assert path.read_bytes() == raw
+ assert backup.read_bytes() == b"older backup"
+ assert dialogs.showerror.call_count == 1
+ dialogs.showinfo.assert_not_called()
+ assert not list(tmp_path.glob(".*.autofix-*"))
+
+
+def test_concurrently_created_backup_is_preserved(monkeypatch, tmp_path):
+ path = tmp_path / "source.py"
+ raw = b"import os\nanswer = 42\n"
+ dialogs = prepare(monkeypatch, path, raw)
+ real_link = m.os.link
+ backup = path.with_suffix(".py.bak")
+ def competing_link(source, target):
+ if target == backup:
+ backup.write_bytes(b"other writer")
+ real_link(source, target)
+ monkeypatch.setattr(m.os, "link", competing_link)
+ m.auto_fix_unused_imports(Mock())
+ dialogs.showerror.assert_not_called()
+ assert backup.read_bytes() == b"other writer"
+ assert path.with_suffix(".py.bak.1").read_bytes() == raw
+ assert path.read_bytes() == b"answer = 42\n"
+
+
+def test_detected_source_change_is_not_overwritten(monkeypatch, tmp_path):
+ path = tmp_path / "source.py"
+ raw = b"import os\nanswer = 42\n"
+ newer = b"import os\nprint(os.getcwd())\n"
+ dialogs = prepare(monkeypatch, path, raw)
+ real_link = m.os.link
+ def edit_during_backup(source, target):
+ real_link(source, target)
+ path.write_bytes(newer)
+ monkeypatch.setattr(m.os, "link", edit_during_backup)
+ m.auto_fix_unused_imports(Mock())
+ assert path.read_bytes() == newer
+ assert path.with_suffix(".py.bak").read_bytes() == raw
+ assert dialogs.showerror.call_count == 1
+ dialogs.showinfo.assert_not_called()
+ assert not list(tmp_path.glob(".*.autofix-*"))
+
+
+def test_success_preserves_file_mode(monkeypatch, tmp_path):
+ path = tmp_path / "source.py"
+ dialogs = prepare(monkeypatch, path, b"import os\nanswer = 42\n")
+ path.chmod(0o755)
+ original_mode = stat.S_IMODE(path.stat().st_mode)
+ m.auto_fix_unused_imports(Mock())
+ dialogs.showerror.assert_not_called()
+ assert stat.S_IMODE(path.stat().st_mode) == original_mode
+
+
+def test_removing_only_import_can_publish_empty_file(monkeypatch, tmp_path):
+ path = tmp_path / "source.py"
+ raw = b"import os\n"
+ dialogs = prepare(monkeypatch, path, raw)
+ m.auto_fix_unused_imports(Mock())
+ dialogs.showerror.assert_not_called()
+ assert path.read_bytes() == b""
+ assert path.with_suffix(".py.bak").read_bytes() == raw
+
+
+def test_mixed_newlines_preserve_untouched_bytes(monkeypatch, tmp_path):
+ path = tmp_path / "source.py"
+ raw = b"import os\r\nanswer = 42\nlabel = 'ok'\r"
+ dialogs = prepare(monkeypatch, path, raw)
+ m.auto_fix_unused_imports(Mock())
+ dialogs.showerror.assert_not_called()
+ assert path.read_bytes() == b"answer = 42\nlabel = 'ok'\r"
+ assert path.with_suffix(".py.bak").read_bytes() == raw