Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .github/workflows/source-platform-smoke.yml
Original file line number Diff line number Diff line change
Expand Up @@ -44,3 +44,8 @@ jobs:
env:
PYTHONIOENCODING: utf-8
run: python -m pytest tests/source_platform_smoke.py tests/test_metadata.py -v --basetemp=.pytest_tmp

- name: Run full regression suite
env:
PYTHONIOENCODING: utf-8
run: python -m pytest tests -v --basetemp=.pytest_tmp
23 changes: 23 additions & 0 deletions EXPORTFORMAT.md
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,29 @@ Stand: 2026-05-25
- Der Export ist offline-first und enthält keine Cloud-Synchronisation.
- BLOB-Werte werden JSON-kompatibel als Base64-Struktur serialisiert.

## Sichere Veröffentlichung von CSV und JSON

Der Export hält sichtbare Daten und Herkunft vor dem Dateidialog fest. Als Ziel
sind die offene Hauptdatenbank, per SQL eingebundene Datenbanken und deren
Standardbegleitdateien `-wal`, `-shm` und `-journal` gesperrt. Das gilt auch für
Dateialiase und für Begleitdateien, die noch nicht existieren. Ein Wechsel oder
Schließen der Datenbank während des Dialogs hebt den ursprünglichen Schutz nicht
auf; neu eingebundene Datenbanken werden zusätzlich berücksichtigt.
Mehrdeutige Windows-Pfadkomponenten mit abschließendem Punkt oder Leerzeichen
werden abgewiesen, bevor Windows den Zielnamen beim Schreiben normalisiert.

Die Ausgabe wird zunächst vollständig in eine eigene temporäre Datei im
Zielverzeichnis geschrieben, synchronisiert und geschlossen. Erst danach wird
das Ziel nach erneuter Schutzprüfung ersetzt. Serialisierungs-, Schreib- oder
Ersetzungsfehler erhalten eine vorhandene Ausgabe; fremde temporäre Dateien
werden nicht entfernt. Fehler bei der Prüfung aktiver Datenbanken brechen den
Export ab.

Diese Prüfungen garantieren keine Transaktion gegen gleichzeitige externe
Dateisystemänderungen zwischen letzter Prüfung und Ersetzung und keine
Stromausfall-Dauerhaftigkeit. Besondere SQLite-VFS-/Superjournal-Dateien und
App-Einstellungsdateien sind nicht Teil dieses Datenbankschutzvertrags.

## Struktur

```json
Expand Down
23 changes: 20 additions & 3 deletions SQLiteViewer.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
from datetime import datetime
from typing import List, Tuple, Any
from translator import TranslationSystem
from export_atomic import atomic_export, database_files

APP_TITLE = "SQLite Viewer Pro"
APP_VERSION = "2.1.0"
Expand Down Expand Up @@ -992,6 +993,12 @@ def export_csv(self):
messagebox.showwarning("Export", action_state["empty_message"])
return

try:
protected = database_files(self)
except Exception as e:
messagebox.showerror("Export-Fehler", str(e))
return

table = export_context.get("table") or self.table_var.get() or export_context.get("view") or "export"
default_name = f"{table}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"

Expand All @@ -1006,7 +1013,9 @@ def export_csv(self):
return

try:
with open(path, "w", newline="", encoding="utf-8-sig") as f:
protected |= database_files(self)
with atomic_export(path, protected, newline="", encoding="utf-8-sig",
refresh_sources=lambda: database_files(self)) as f:
writer = csv.writer(f, delimiter=";", quoting=csv.QUOTE_MINIMAL)
writer.writerow(columns)
# Bugsweep 23: BLOB/bytes base64-kodieren, sonst landet ein b'...'-Rohliteral in der
Expand Down Expand Up @@ -1085,6 +1094,13 @@ def export_json(self):
messagebox.showwarning("Export", action_state["empty_message"])
return

try:
protected = database_files(self)
payload = self._build_export_payload()
except Exception as e:
messagebox.showerror("Export-Fehler", str(e))
return

table = export_context.get("table") or self.table_var.get() or export_context.get("view") or "export"
default_name = f"{table}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
path = filedialog.asksaveasfilename(
Expand All @@ -1098,8 +1114,9 @@ def export_json(self):
return

try:
payload = self._build_export_payload()
with open(path, "w", encoding="utf-8") as handle:
protected |= database_files(self)
with atomic_export(path, protected, encoding="utf-8",
refresh_sources=lambda: database_files(self)) as handle:
json.dump(payload, handle, indent=2, ensure_ascii=False)

self._set_status(f"Exportiert: {os.path.basename(path)}")
Expand Down
77 changes: 77 additions & 0 deletions export_atomic.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
"""Failure-preserving text exports with SQLite source-file protection."""
from contextlib import contextmanager
import logging
import os
from pathlib import Path
import tempfile


def database_files(viewer):
"""Capture main and attached database names, including SQLite sidecars."""
names = []
if getattr(viewer, 'db_path', None):
names.append(viewer.db_path)
connection = getattr(viewer, 'conn', None)
if connection is not None:
cursor = connection.execute('PRAGMA database_list')
try:
names.extend(row[2] for row in cursor.fetchall() if row[2])
finally:
cursor.close()
paths = set()
for name in names:
if name == ':memory:':
continue
for path in (Path(name).absolute(), Path(name).resolve()):
paths.add(path)
paths.update(Path(str(path) + suffix) for suffix in ('-wal', '-shm', '-journal'))
return paths


def protect_destination(destination, sources):
lexical_target = Path(os.path.abspath(destination))
if os.name == 'nt' and any(
part not in ('.', '..') and part.endswith((' ', '.'))
for part in Path(destination).parts
):
raise ValueError('Das Exportziel enthält einen mehrdeutigen Windows-Dateinamen. Bitte einen anderen Pfad wählen.')
target = Path(destination).resolve()
for source in sources:
if lexical_target == Path(os.path.abspath(source)) or target == source.resolve():
raise ValueError('Das Exportziel ist eine geschützte Datenbankdatei. Bitte einen anderen Pfad wählen.')
try:
same = os.path.samefile(destination, source)
except FileNotFoundError:
same = False
if same:
raise ValueError('Das Exportziel ist eine geschützte Datenbankdatei. Bitte einen anderen Pfad wählen.')


@contextmanager
def atomic_export(destination, sources, *, encoding, newline=None, refresh_sources=None):
"""Publish a complete sibling temporary file after rechecking identity."""
destination = Path(destination).absolute()
protect_destination(destination, sources)
temporary = None
try:
with tempfile.NamedTemporaryFile(mode='w', encoding=encoding, newline=newline,
dir=destination.absolute().parent,
prefix='.sqliteviewer-export-', suffix='.tmp',
delete=False) as handle:
temporary = Path(handle.name)
yield handle
handle.flush()
os.fsync(handle.fileno())
if refresh_sources is not None:
sources = sources | refresh_sources()
protect_destination(destination, sources)
os.replace(temporary, destination)
temporary = None
finally:
if temporary is not None:
try:
temporary.unlink()
except FileNotFoundError:
pass
except OSError:
logging.getLogger(__name__).warning('Cannot remove own export temporary file: %s', temporary, exc_info=True)
Loading
Loading