diff --git a/steward/backup/check.py b/steward/backup/check.py index a74b807..118e6ba 100644 --- a/steward/backup/check.py +++ b/steward/backup/check.py @@ -36,15 +36,14 @@ from sqlalchemy.ext.asyncio import create_async_engine from .contents import (ACTIVITY, ACTIVITY_EVERYWHERE, AREAS, EXCLUDED, HAND_SET_ONLY, STATE_SETTINGS) from .dump import BackupError, read_backup -from .restore import _async_url +from .pgdata import async_url as _async_url +from .pgdata import copy_rows, restore_script, script_tables -CHECK_DEADLINE_SECONDS = 5 * 60 # Names shown for a row, the first a table has; else its primary key. LABEL_COLUMNS = ("name", "username", "key", "title", "widget_key", "label", "mac") # A setting's two values are shown when they fit; longer ones are cut. SHOWN_CHARS = 120 _TOKEN = re.compile(r"enc:v1:[A-Za-z0-9_\-=]+") -_COPY = re.compile(r'^COPY ([^ ]+) \((.*)\) FROM stdin;$') @dataclass @@ -134,79 +133,6 @@ class CheckReport: return {"ok": self.ok, "changes": self.changes, "rows": self.rows, "summary": self.summary()} -# ── COPY text ─────────────────────────────────────────────────────────────── - -_ESCAPES = {ord("b"): b"\b", ord("f"): b"\f", ord("n"): b"\n", ord("r"): b"\r", - ord("t"): b"\t", ord("v"): b"\v"} - - -def unescape(field: bytes) -> str | None: - """One field of COPY's text format: \\N is NULL, backslash escapes are - undone (named, octal \\ooo and hex \\xhh).""" - if field == b"\\N": - return None - if b"\\" not in field: - return field.decode("utf-8", errors="replace") - out = bytearray() - i, n = 0, len(field) - while i < n: - c = field[i] - if c != 0x5C or i + 1 == n: - out.append(c) - i += 1 - continue - nxt = field[i + 1] - if nxt in _ESCAPES: - out += _ESCAPES[nxt] - i += 2 - elif 0x30 <= nxt <= 0x37: - j = i + 1 - while j < n and j < i + 4 and 0x30 <= field[j] <= 0x37: - j += 1 - out.append(int(field[i + 1:j], 8) & 0xFF) - i = j - elif nxt == ord("x") and i + 2 < n and chr(field[i + 2]) in "0123456789abcdefABCDEF": - j = i + 2 - while j < n and j < i + 4 and chr(field[j]) in "0123456789abcdefABCDEF": - j += 1 - out.append(int(field[i + 2:j], 16)) - i = j - else: - out.append(nxt) - i += 2 - return bytes(out).decode("utf-8", errors="replace") - - -def copy_rows(data: bytes) -> list[list[str | None]]: - """COPY text rows (no header, no terminator) as lists of fields.""" - return [[unescape(f) for f in line.split(b"\t")] - for line in data.split(b"\n") if line and line != b"\\."] - - -def _ident(name: str) -> str: - name = name.strip() - return name[1:-1].replace('""', '"') if name.startswith('"') and name.endswith('"') else name - - -def script_tables(script: bytes) -> dict[str, tuple[list[str], list[list[str | None]]]]: - """Each table's (columns, rows) from a pg_restore --data-only script.""" - out: dict[str, tuple[list[str], list[list[str | None]]]] = {} - lines = iter(script.split(b"\n")) - for line in lines: - m = _COPY.match(line.decode("utf-8", errors="replace")) - if not m: - continue - table = _ident(m.group(1).split(".")[-1]) - columns = [_ident(c) for c in m.group(2).split(",")] - body = [] - for row in lines: - if row == b"\\.": - break - body.append(row) - out[table] = (columns, copy_rows(b"\n".join(body))) - return out - - # ── Comparing ─────────────────────────────────────────────────────────────── def _fernet(key: str): @@ -355,26 +281,8 @@ def _labels(tables: dict[str, _Side]) -> dict[str, dict[str, str]]: # ── Reading both sides ────────────────────────────────────────────────────── async def _backup_rows(dump: Path) -> dict[str, tuple[list[str], list[list[str | None]]]]: - """Every table's rows in the dump, as pg_restore would load them. Run - without a database: pg_restore writes the script and nothing connects.""" - try: - proc = await asyncio.create_subprocess_exec( - "pg_restore", "--data-only", "--no-owner", "--no-privileges", "--file=-", str(dump), - stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) - except FileNotFoundError: - raise BackupError("pg_restore isn't installed here.") from None - try: - out, err = await asyncio.wait_for(proc.communicate(), CHECK_DEADLINE_SECONDS) - except TimeoutError: - proc.kill() - await proc.wait() - raise BackupError(f"Reading the backup didn't finish within " - f"{CHECK_DEADLINE_SECONDS // 60} minutes.") from None - if proc.returncode != 0: - lines = [ln for ln in err.decode(errors="replace").splitlines() if ln.strip()] - raise BackupError("The backup can't be read: " + (lines[-1] if lines else - f"pg_restore exited {proc.returncode}.")) - return script_tables(out) + """Every table's rows in the dump, as pg_restore would load them.""" + return script_tables(await restore_script(dump)) async def _here(database_url: str, tables: list[str]): diff --git a/steward/backup/pgdata.py b/steward/backup/pgdata.py new file mode 100644 index 0000000..b6be616 --- /dev/null +++ b/steward/backup/pgdata.py @@ -0,0 +1,131 @@ +"""A backup's rows as COPY text: read out of the dump by pg_restore, without a +database, and parsed or handed on as they are. The check (check.py) parses +them to compare; the in-place restore (restore.py) hands them to COPY +unparsed, so what goes back in is byte for byte what the dump holds. +""" +from __future__ import annotations + +import asyncio +import re +from pathlib import Path + +from .dump import BackupError + +SCRIPT_DEADLINE_SECONDS = 5 * 60 +_COPY = re.compile(r'^COPY ([^ ]+) \((.*)\) FROM stdin;$') +_ESCAPES = {ord("b"): b"\b", ord("f"): b"\f", ord("n"): b"\n", ord("r"): b"\r", + ord("t"): b"\t", ord("v"): b"\v"} +_HEX = "0123456789abcdefABCDEF" + + +def async_url(database_url: str) -> str: + from sqlalchemy.engine import make_url + url = make_url(database_url) + return str(url.set(drivername="postgresql+asyncpg")) if url.drivername == "postgresql" else database_url + + +async def restore_script(dump: Path) -> bytes: + """The dump's rows as pg_restore would load them: its --data-only script. + Nothing connects; pg_restore only writes the script.""" + try: + proc = await asyncio.create_subprocess_exec( + "pg_restore", "--data-only", "--no-owner", "--no-privileges", "--file=-", str(dump), + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) + except FileNotFoundError: + raise BackupError("pg_restore isn't installed here.") from None + try: + out, err = await asyncio.wait_for(proc.communicate(), SCRIPT_DEADLINE_SECONDS) + except TimeoutError: + proc.kill() + await proc.wait() + raise BackupError(f"Reading the backup didn't finish within " + f"{SCRIPT_DEADLINE_SECONDS // 60} minutes.") from None + if proc.returncode != 0: + lines = [ln for ln in err.decode(errors="replace").splitlines() if ln.strip()] + raise BackupError("The backup can't be read: " + (lines[-1] if lines else + f"pg_restore exited {proc.returncode}.")) + return out + + +def _ident(name: str) -> str: + name = name.strip() + return name[1:-1].replace('""', '"') if name.startswith('"') and name.endswith('"') else name + + +def script_blocks(script: bytes) -> dict[str, tuple[list[str], bytes]]: + """Each table's (columns, COPY text) from a --data-only script, the text + as it stands: one line per row, ready for COPY … FROM in text format.""" + out: dict[str, tuple[list[str], bytes]] = {} + lines = iter(script.split(b"\n")) + for line in lines: + m = _COPY.match(line.decode("utf-8", errors="replace")) + if not m: + continue + table = _ident(m.group(1).split(".")[-1]) + columns = [_ident(c) for c in m.group(2).split(",")] + body = [] + for row in lines: + if row == b"\\.": + break + body.append(row + b"\n") + out[table] = (columns, b"".join(body)) + return out + + +def unescape(field: bytes) -> str | None: + """One field of COPY's text format: \\N is NULL, backslash escapes are + undone (named, octal \\ooo and hex \\xhh).""" + if field == b"\\N": + return None + if b"\\" not in field: + return field.decode("utf-8", errors="replace") + out = bytearray() + i, n = 0, len(field) + while i < n: + c = field[i] + if c != 0x5C or i + 1 == n: + out.append(c) + i += 1 + continue + nxt = field[i + 1] + if nxt in _ESCAPES: + out += _ESCAPES[nxt] + i += 2 + elif 0x30 <= nxt <= 0x37: + j = i + 1 + while j < n and j < i + 4 and 0x30 <= field[j] <= 0x37: + j += 1 + out.append(int(field[i + 1:j], 8) & 0xFF) + i = j + elif nxt == ord("x") and i + 2 < n and chr(field[i + 2]) in _HEX: + j = i + 2 + while j < n and j < i + 4 and chr(field[j]) in _HEX: + j += 1 + out.append(int(field[i + 2:j], 16)) + i = j + else: + out.append(nxt) + i += 2 + return bytes(out).decode("utf-8", errors="replace") + + +def copy_rows(data: bytes) -> list[list[str | None]]: + """COPY text rows (no header, no terminator) as lists of fields.""" + return [[unescape(f) for f in line.split(b"\t")] + for line in data.split(b"\n") if line and line != b"\\."] + + +def script_tables(script: bytes) -> dict[str, tuple[list[str], list[list[str | None]]]]: + """Each table's (columns, rows) from a --data-only script.""" + return {t: (cols, copy_rows(raw)) for t, (cols, raw) in script_blocks(script).items()} + + +def script_from(blocks: dict[str, tuple[list[str], bytes]]) -> bytes: + """A --data-only script holding `blocks`: the shape pg_restore writes, + so a test can stand in for a dump.""" + out = [] + for table, (columns, raw) in blocks.items(): + out.append(f"COPY public.{table} ({', '.join(columns)}) FROM stdin;\n".encode()) + out.append(raw) + out.append(b"\\.\n\n") + return b"".join(out) diff --git a/steward/backup/restore.py b/steward/backup/restore.py index cfe1b11..60f094f 100644 --- a/steward/backup/restore.py +++ b/steward/backup/restore.py @@ -1,4 +1,6 @@ -"""Bringing Steward back from a backup file into an empty database. +"""Bringing Steward back from a backup file: into an empty database +(`restore`, the CLI and first-run setup), or in place on a running Steward +(`restore_in_place`, below). pg_restore runs in its three sections so the rows can be put right between the data and the constraints: a kept table may point at a row of a table @@ -13,6 +15,7 @@ always does, so a backup made by an older Steward comes up to date. from __future__ import annotations import asyncio +import io import logging import tempfile from dataclasses import dataclass @@ -23,6 +26,7 @@ from sqlalchemy.ext.asyncio import create_async_engine from .contents import EXCLUDED from .dump import BackupError, pg_env, read_backup +from .pgdata import async_url as _async_url logger = logging.getLogger(__name__) @@ -147,12 +151,6 @@ async def _pg_restore(database_url: str, dump: Path, section: str) -> None: raise BackupError(f"Restoring {section}: " + (lines[-1] if lines else f"{name} exited {proc.returncode}.")) -def _async_url(database_url: str) -> str: - from sqlalchemy.engine import make_url - url = make_url(database_url) - return str(url.set(drivername="postgresql+asyncpg")) if url.drivername == "postgresql" else database_url - - async def _tables(database_url: str) -> list[str]: engine = create_async_engine(_async_url(database_url)) try: @@ -195,3 +193,223 @@ async def restore(path: Path, database_url: str, *, wipe: bool = False, fixed = await _execute(database_url, dangling_fixes(metadata or all_metadata(plugin_dirs))) await _pg_restore(database_url, dump, "post-data") return Restored(info, key, fixed) + + +# ── In place, on a running Steward (#5815) ─────────────────────────────────── +# A live restore replaces what the backup carries and keeps everything else: +# the audit log, alert history and metrics stay. It runs in one transaction, +# so a failure anywhere leaves the database as it was: +# 1. keep what Steward records as it runs (contents.ACTIVITY on rows that +# come back, contents.STATE_SETTINGS) aside in temporary tables; +# 2. drop every foreign key touching a replaced table; +# 3. empty the replaced tables and COPY the backup's rows in, unparsed; +# 4. put back what was kept aside, and move each sequence past its rows; +# 5. put right the rows whose parent didn't come back, by their foreign +# key: CASCADE removes the row, SET NULL clears the link, and a RESTRICT +# or NO ACTION link is cleared when it may be empty (a run whose user is +# gone keeps its record) and removed when it may not; +# 6. add the foreign keys back, which checks every row. +# Steward's database user owns its tables, so none of this needs a superuser. + + +class NotSameSchema(BackupError): + """The backup was made at another schema: in place won't do.""" + + def __init__(self, schema: str, backup_heads: list[str], here_heads: list[str]): + self.schema = schema + if schema == "newer": + said = "It was made by a newer Steward than this one; update Steward first." + else: + said = ("It was made by an older Steward, so it can't go back in place; restoring it " + "replaces everything, history included.") + super().__init__(said) + self.backup_heads, self.here_heads = backup_heads, here_heads + + +@dataclass +class RestoredInPlace: + manifest: dict + secret_key: str + tables: list[str] + rows: int + fixed: int + + +@dataclass +class _ForeignKey: + name: str + child: str + parent: str + columns: list[str] + parent_columns: list[str] + on_delete: str # pg_constraint.confdeltype: c, n, a, r, d + nullable: bool + definition: str + + +def _q(name: str) -> str: + return '"' + name.replace('"', '""') + '"' + + +def fk_fix(fk: _ForeignKey) -> str | None: + """The statement putting right `fk`'s rows whose parent is gone; None for + a key over several columns (re-adding it still checks every row).""" + if len(fk.columns) != 1: + return None + col, pcol = _q(fk.columns[0]), _q(fk.parent_columns[0]) + gone = (f"x.{col} IS NOT NULL AND NOT EXISTS " + f"(SELECT 1 FROM public.{_q(fk.parent)} p WHERE p.{pcol} = x.{col})") + if fk.on_delete == "n" or (fk.on_delete in ("a", "r") and fk.nullable): + return f"UPDATE public.{_q(fk.child)} x SET {col} = NULL WHERE {gone}" + return f"DELETE FROM public.{_q(fk.child)} x WHERE {gone}" + + +async def _layout(conn, tables: list[str]): + """What the transaction needs to know first: the foreign keys touching + `tables`, each table's primary key and columns, and its sequences.""" + fks = [] + for row in (await conn.execute(text( + "SELECT con.conname, cl.relname, ref.relname, con.confdeltype, " + "pg_get_constraintdef(con.oid), " + "ARRAY(SELECT a.attname FROM unnest(con.conkey) WITH ORDINALITY k(n, i) " + " JOIN pg_attribute a ON a.attrelid = con.conrelid AND a.attnum = k.n ORDER BY k.i), " + "ARRAY(SELECT a.attname FROM unnest(con.confkey) WITH ORDINALITY k(n, i) " + " JOIN pg_attribute a ON a.attrelid = con.confrelid AND a.attnum = k.n ORDER BY k.i), " + "NOT bool_or_notnull.v " + "FROM pg_constraint con " + "JOIN pg_class cl ON cl.oid = con.conrelid " + "JOIN pg_class ref ON ref.oid = con.confrelid " + "JOIN pg_namespace ns ON ns.oid = cl.relnamespace " + "CROSS JOIN LATERAL (SELECT bool_or(a.attnotnull) AS v FROM pg_attribute a " + " WHERE a.attrelid = con.conrelid AND a.attnum = ANY(con.conkey)) bool_or_notnull " + "WHERE con.contype = 'f' AND ns.nspname = 'public' " + "AND (cl.relname = ANY(:t) OR ref.relname = ANY(:t)) ORDER BY con.conname"), + {"t": tables})).all(): + fks.append(_ForeignKey(row[0], row[1], row[2], list(row[5]), list(row[6]), row[3], + bool(row[7]), row[4])) + keys: dict[str, list[str]] = {} + for t, c in (await conn.execute(text( + "SELECT cl.relname, a.attname FROM pg_index i " + "JOIN pg_class cl ON cl.oid = i.indrelid " + "JOIN pg_namespace ns ON ns.oid = cl.relnamespace " + "JOIN pg_attribute a ON a.attrelid = cl.oid AND a.attnum = ANY(i.indkey) " + "WHERE i.indisprimary AND ns.nspname = 'public' AND cl.relname = ANY(:t) " + "ORDER BY cl.relname, a.attnum"), {"t": tables})).all(): + keys.setdefault(t, []).append(c) + columns: dict[str, list[str]] = {} + sequences: list[tuple[str, str, str]] = [] + for t, c, default in (await conn.execute(text( + "SELECT table_name, column_name, column_default FROM information_schema.columns " + "WHERE table_schema = 'public' AND table_name = ANY(:t) " + "ORDER BY table_name, ordinal_position"), {"t": tables})).all(): + columns.setdefault(t, []).append(c) + if default and default.startswith("nextval("): + seq = (await conn.execute(text("SELECT pg_get_serial_sequence(:t, :c)"), + {"t": f"public.{_q(t)}", "c": c})).scalar() + if seq: + sequences.append((t, c, seq)) + return fks, keys, columns, sequences + + +async def restore_in_place(path: Path, database_url: str, *, + plugin_dirs: list[Path] | None = None) -> RestoredInPlace: + """Put the backup at `path` back into the running Steward's database, + replacing the tables it carries and keeping the rest (see above). Raises + NotSameSchema for a backup made at another schema, and BackupError for + one that can't be read or doesn't fit; the database is then untouched.""" + from .check import known_revisions, schema_of + from .contents import ACTIVITY, STATE_SETTINGS + from .pgdata import restore_script, script_blocks + + with tempfile.TemporaryDirectory() as tmp: + info, dump, key = await asyncio.to_thread(read_backup, path, Path(tmp)) + blocks = script_blocks(await restore_script(dump)) + engine = create_async_engine(_async_url(database_url)) + try: + async with engine.connect() as conn: + heads = sorted((await conn.execute(text("SELECT version_num FROM alembic_version"))).scalars()) + backup_heads = sorted(info.get("migration_heads") or []) + schema = schema_of(backup_heads, heads, known_revisions(plugin_dirs)) + if schema != "same": + raise NotSameSchema(schema, backup_heads, heads) + present = set((await conn.execute(text( + "SELECT tablename FROM pg_tables WHERE schemaname = 'public'"))).scalars()) + tables = sorted(t for t in blocks if t not in EXCLUDED and t in present) + fks, keys, columns, sequences = await _layout(conn, tables) + for t in tables: + extra = set(blocks[t][0]) - set(columns.get(t, [])) + if extra: + raise BackupError(f"The backup's {t} has columns this Steward lacks: " + f"{', '.join(sorted(extra))}.") + + rows = sum(raw.count(b"\n") for _, raw in (blocks[t] for t in tables)) + try: + async with engine.begin() as conn: + # 1. What Steward records as it runs, kept aside. + if "app_settings" in tables: + await conn.execute(text( + "CREATE TEMP TABLE _keep_settings ON COMMIT DROP AS " + "SELECT * FROM public.app_settings WHERE key = ANY(:k)"), + {"k": sorted(STATE_SETTINGS)}) + kept_aside: dict[str, list[str]] = {} + for t in tables: + act = [c for c in columns.get(t, []) if c in ACTIVITY.get(t, ())] + if act and keys.get(t): + kept_aside[t] = act + cols = ", ".join(_q(c) for c in keys[t] + act) + await conn.execute(text( + f"CREATE TEMP TABLE {_q('_keep_' + t)} ON COMMIT DROP AS " + f"SELECT {cols} FROM public.{_q(t)}")) + # 2. The foreign keys touching what is replaced. + for fk in fks: + await conn.execute(text( + f"ALTER TABLE public.{_q(fk.child)} DROP CONSTRAINT {_q(fk.name)}")) + # 3. The backup's rows, as the dump holds them. + await conn.execute(text("TRUNCATE " + ", ".join(f"public.{_q(t)}" for t in tables))) + raw = (await conn.get_raw_connection()).driver_connection + for t in tables: + cols, data = blocks[t] + if data: + await raw.copy_to_table(t, schema_name="public", source=io.BytesIO(data), + columns=cols, format="text") + # 4. Put back what was kept aside; sequences past their rows. + if "app_settings" in tables: + await conn.execute(text("DELETE FROM public.app_settings WHERE key = ANY(:k)"), + {"k": sorted(STATE_SETTINGS)}) + await conn.execute(text("INSERT INTO public.app_settings SELECT * FROM _keep_settings")) + for t, act in kept_aside.items(): + sets = ", ".join(f"{_q(c)} = k.{_q(c)}" for c in act) + match = " AND ".join(f"x.{_q(c)} = k.{_q(c)}" for c in keys[t]) + await conn.execute(text( + f"UPDATE public.{_q(t)} x SET {sets} FROM {_q('_keep_' + t)} k WHERE {match}")) + for t, c, seq in sequences: + # `seq` is pg_get_serial_sequence's own answer, already quoted. + literal = "'" + seq.replace("'", "''") + "'" + await conn.execute(text( + f"SELECT setval({literal}, GREATEST(COALESCE((SELECT max({_q(c)}) FROM public.{_q(t)}), 0), " + f"(SELECT last_value FROM {seq}), 1))")) + # 5. Rows whose parent didn't come back. A removal can leave + # another row without its parent, so until nothing moves. + fixes = [sql for sql in (fk_fix(fk) for fk in fks) if sql] + fixed = 0 + for _ in range(5): + moved = 0 + for sql in fixes: + moved += max((await conn.execute(text(sql))).rowcount or 0, 0) + fixed += moved + if not moved: + break + # 6. The keys back on, each row checked. + for fk in fks: + await conn.execute(text( + f"ALTER TABLE public.{_q(fk.child)} ADD CONSTRAINT {_q(fk.name)} {fk.definition}")) + except BackupError: + raise + except Exception as exc: + said = getattr(exc, "orig", None) or exc + raise BackupError(f"The restore was undone, nothing changed: {said}") from exc + finally: + await engine.dispose() + logger.warning("Restore in place: replaced %d tables (%d rows) from %s; %d rows put right", + len(tables), rows, path.name, fixed) + return RestoredInPlace(info, key, tables, rows, fixed) diff --git a/tests/core/test_backup_check.py b/tests/core/test_backup_check.py index 000552b..59242cf 100644 --- a/tests/core/test_backup_check.py +++ b/tests/core/test_backup_check.py @@ -8,8 +8,9 @@ from quart import render_template from steward.backup import check from steward.backup.check import (AreaCheck, CheckReport, RowDiff, TableCheck, _Opener, _Side, - compare_table, copy_rows, normalise, schema_of, script_tables, - unescape) + compare_table, normalise, schema_of) +from steward.backup.pgdata import (copy_rows, script_blocks, script_from, script_tables, + unescape) from steward.core.crypto import _make_fernet from tests.test_style_tokens import hex_colours @@ -48,6 +49,9 @@ def test_a_restore_script_gives_each_tables_rows(): assert tables["app_settings"] == (["key", "value_json", "updated_at"], [["smtp.port", "587", "2026-10-15 14:00:00+00"]]) assert tables["audit_events"][0] == ["id", "timestamp", "action"] + # Handed on unparsed, a block is the COPY text as the dump holds it. + assert script_blocks(script)["app_settings"][1] == b"smtp.port\t587\t2026-10-15 14:00:00+00\n" + assert script_tables(script_from(script_blocks(script))) == tables def test_an_instant_is_the_same_whatever_zone_prints_it(): diff --git a/tests/core/test_backup_restore_in_place.py b/tests/core/test_backup_restore_in_place.py new file mode 100644 index 0000000..7a7acec --- /dev/null +++ b/tests/core/test_backup_restore_in_place.py @@ -0,0 +1,32 @@ +"""How an in-place restore puts right a row whose parent didn't come back +(steward/backup/restore.py, #5815): by the row's own foreign key.""" +from steward.backup.restore import NotSameSchema, _ForeignKey, fk_fix + + +def _fk(on_delete, nullable=True, columns=("rule_id",)): + return _ForeignKey("fk", "alert_events", "alert_rules", list(columns), ["id"] * len(columns), + on_delete, nullable, "FOREIGN KEY (rule_id) REFERENCES alert_rules(id)") + + +def test_a_cascading_key_removes_the_row(): + assert fk_fix(_fk("c")) == ( + 'DELETE FROM public."alert_events" x WHERE x."rule_id" IS NOT NULL AND NOT EXISTS ' + '(SELECT 1 FROM public."alert_rules" p WHERE p."id" = x."rule_id")') + + +def test_set_null_and_a_restricting_key_that_may_be_empty_clear_the_link(): + for action in ("n", "a", "r"): + assert fk_fix(_fk(action)).startswith('UPDATE public."alert_events" x SET "rule_id" = NULL') + + +def test_a_restricting_key_that_may_not_be_empty_removes_the_row(): + assert fk_fix(_fk("r", nullable=False)).startswith('DELETE FROM public."alert_events"') + + +def test_a_key_over_several_columns_is_left_to_the_constraint(): + assert fk_fix(_fk("c", columns=("a", "b"))) is None + + +def test_another_schema_says_what_to_do(): + assert "update Steward first" in str(NotSameSchema("newer", ["x"], ["y"])) + assert "replaces everything" in str(NotSameSchema("older", ["x"], ["y"])) diff --git a/tests/integration/test_backup_restore_in_place.py b/tests/integration/test_backup_restore_in_place.py new file mode 100644 index 0000000..744bda5 --- /dev/null +++ b/tests/integration/test_backup_restore_in_place.py @@ -0,0 +1,177 @@ +"""Integration: a backup put back in place on a running Steward (#5815). It +replaces what the backup carries and keeps the rest: settings, hosts and +rules come back as they were; the audit log and the history of what is +still there stay; history of what didn't come back goes with it. A failure +part-way changes nothing. pg_restore isn't in this lane, so the backup's +rows are this database's own, read out as COPY text before the changes. +Requires STEWARD_DATABASE_URL.""" +from __future__ import annotations + +import asyncio +import io +import os +import uuid +from datetime import datetime, timezone + +import pytest + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +async def _snapshot(app) -> bytes: + """The kept tables as they stand, as a --data-only script.""" + from sqlalchemy import text + from steward.backup import contents + from steward.backup.pgdata import script_from + blocks = {} + async with app._db_engine.connect() as conn: + order: dict[str, list[str]] = {} + for t, c in (await conn.execute(text( + "SELECT table_name, column_name FROM information_schema.columns " + "WHERE table_schema = 'public' AND table_name = ANY(:t) " + "ORDER BY table_name, ordinal_position"), {"t": sorted(contents.KEPT)})).all(): + order.setdefault(t, []).append(c) + raw = (await conn.get_raw_connection()).driver_connection + for t, cols in order.items(): + buf = io.BytesIO() + await raw.copy_from_table(t, columns=cols, output=buf, format="text") + blocks[t] = (cols, buf.getvalue()) + return script_from(blocks) + + +async def _backup_file(app, tmp_path, heads=None): + from steward.backup import dump, job + (tmp_path / "steward.dump").write_bytes(b"PGDMP") + path = tmp_path / "steward-20261015T030000Z.tar" + heads = heads if heads is not None else await job.migration_heads(app.config["DATABASE_URL"]) + dump.pack(path, tmp_path / "steward.dump", app.config["SECRET_KEY"], + dump.manifest(datetime.now(timezone.utc), heads)) + return path + + +def test_a_backup_goes_back_in_place_and_history_stays(app, monkeypatch, tmp_path): + from sqlalchemy import func, select + from steward.backup import pgdata + from steward.backup.restore import restore_in_place + from steward.core.settings import get_all_settings, set_setting + from steward.models.alerts import AlertEvent, AlertOperator, AlertRule + from steward.models.audit import AuditEvent + from steward.models.hosts import Host + from steward.models.users import User, UserRole + + tag = uuid.uuid4().hex[:8] + setting = f"test.in_place_{tag}" + + def rule(name, uid): + return AlertRule(name=name, source_module="host_agent", resource_name=f"r-{tag}", + metric_name="cpu_pct", operator=AlertOperator.gt, threshold=90.0, + consecutive_failures_required=1, enabled=True, created_by=uid) + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + user = User(username=f"ip-{tag}", email=f"ip-{tag}@test", password_hash="x", + role=UserRole.admin) + s.add(user) + await s.flush() + await set_setting(s, setting, "before") + kept_host = Host(name=f"kept-{tag}", address="10.7.0.1") + kept_rule = rule(f"kept {tag}", user.id) + s.add_all([kept_host, kept_rule]) + await s.flush() + s.add(AlertEvent(rule_id=kept_rule.id, from_state="inactive", to_state="firing")) + ids = {"user": user.id, "kept_host": kept_host.id, "kept_rule": kept_rule.id} + script = await _snapshot(app) + + async def restore_script(_dump): + return script + monkeypatch.setattr(pgdata, "restore_script", restore_script) + path = await _backup_file(app, tmp_path) + + async with app.db_sessionmaker() as s: + async with s.begin(): + await set_setting(s, setting, "after") + await set_setting(s, "backup.last_attempt_at", f"kept-{tag}") # Steward's own record + new_host = Host(name=f"new-{tag}", address="10.7.0.2") + new_rule = rule(f"new {tag}", ids["user"]) + s.add_all([new_host, new_rule]) + await s.flush() + s.add(AlertEvent(rule_id=new_rule.id, from_state="inactive", to_state="firing")) + s.add(AuditEvent(username="test", action=f"test.{tag}")) + ids |= {"new_host": new_host.id, "new_rule": new_rule.id} + + done = await restore_in_place(path, app.config["DATABASE_URL"]) + + async with app.db_sessionmaker() as s: + settings = await get_all_settings(s) + hosts = set((await s.execute(select(Host.id).where( + Host.id.in_([ids["kept_host"], ids["new_host"]])))).scalars()) + events = dict((await s.execute(select(AlertEvent.rule_id, func.count()).where( + AlertEvent.rule_id.in_([ids["kept_rule"], ids["new_rule"]])) + .group_by(AlertEvent.rule_id))).all()) + audits = (await s.execute(select(func.count()).select_from(AuditEvent).where( + AuditEvent.action == f"test.{tag}"))).scalar_one() + return ids, done, settings, hosts, events, audits + + ids, done, settings, hosts, events, audits = asyncio.run(_go()) + assert settings[setting] == "before" # settings came back + assert settings["backup.last_attempt_at"] == f"kept-{tag}" # Steward's own record stayed + assert hosts == {ids["kept_host"]} # the new host is gone + assert events == {ids["kept_rule"]: 1} # its rule's history went with it + assert audits == 1 # the audit log stayed + assert "app_settings" in done.tables and "audit_events" not in done.tables + assert done.fixed >= 1 + + +def test_a_restore_that_fails_part_way_changes_nothing(app, monkeypatch, tmp_path): + from steward.backup import pgdata + from steward.backup.dump import BackupError + from steward.backup.restore import restore_in_place + from steward.core.settings import get_all_settings, set_setting + + tag = uuid.uuid4().hex[:8] + setting = f"test.undone_{tag}" + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await set_setting(s, setting, "here") + script = await _snapshot(app) + # A monitor row with too few fields: COPY refuses it after the + # settings have already been replaced in the same transaction. + script += b"COPY public.monitors (id, name) FROM stdin;\nonly-one-field\n\\.\n" + + async def restore_script(_dump): + return script + monkeypatch.setattr(pgdata, "restore_script", restore_script) + path = await _backup_file(app, tmp_path) + async with app.db_sessionmaker() as s: + async with s.begin(): + await set_setting(s, setting, "changed") + with pytest.raises(BackupError, match="undone, nothing changed"): + await restore_in_place(path, app.config["DATABASE_URL"]) + async with app.db_sessionmaker() as s: + return (await get_all_settings(s))[setting] + + assert asyncio.run(_go()) == "changed" + + +def test_a_backup_from_another_schema_is_refused_in_place(app, tmp_path): + from steward.backup.restore import NotSameSchema, restore_in_place + + async def _go(): + path = await _backup_file(app, tmp_path, heads=["9999_from_the_future"]) + with pytest.raises(NotSameSchema) as caught: + await restore_in_place(path, app.config["DATABASE_URL"]) + return caught.value + + refused = asyncio.run(_go()) + assert refused.schema == "newer" and "newer Steward" in str(refused)