Harden error output, spool delivery, report files and catalogue loading

README.md

@@ -158,7 +158,7 @@ cloudexit catalogue import FILE.db.gz     # install a file on an air-gapped mach
 cloudexit aws --tfstate infra.tfstate --catalogue FILE.db.gz   # use a file for one run, without installing it
 ```
 
-Set `CLOUDEXIT_CATALOGUE_URL` to download from a mirror instead of the default location. A `./datasets/data.db` from an earlier version is copied to the data directory once.
+Set `CLOUDEXIT_CATALOGUE_URL` to download from a mirror instead of the default location. A `./datasets/data.db` from an earlier version is not picked up on its own; install it on purpose with `cloudexit catalogue import datasets/data.db`. A catalogue that holds a trigger or a view is refused, wherever it comes from.
 
 ## Sync document
 
@@ -209,7 +209,7 @@ The local report is always the Basic report, computed on your machine, whether o
 | --- | --- |
 | 10 | The platform rejected the data, or it is beyond a limit the platform announces. Retrying the same data will not help. |
 | 11 | The API key is missing, mistyped, expired, revoked, not allowed to do this, or belongs to another region. |
-| 12 | The platform is unreachable, busy or in maintenance. Running again is safe: the run keeps the same sync ID. |
+| 12 | The platform is unreachable, busy or in maintenance. A spooled sync is delivered on the next run, and that run then makes a new assessment of its own: see [Retries and the spool](#retries-and-the-spool). |
 | 13 | Credits or a quota ran out. |
 | 14 | The assessment named by `--assessment-id` is missing, deleted, archived, or does not match the data. |
 | 15 | The platform does not accept this cloudexit version or the sync document version. |
@@ -220,7 +220,7 @@ Before anything is sent the key is checked on your machine (prefix, length and c
 
 A failed upload is retried when it may pass: connection errors, timeouts, HTTP 408, 425, 429, 500, 502, 503 and 504, and any error the platform marks as retryable. The wait doubles from 1 second up to 60 seconds with random jitter, `Retry-After` is honoured, and every attempt carries the same idempotency key (the sync ID), so a repeat never stores or charges a second time. All attempts together stay within `--upload-budget` seconds (default 300, `CLOUDEXIT_UPLOAD_BUDGET`). Final failures (exit codes 10, 11, 13, 14 and 15) are not retried.
 
-When the budget runs out on retryable failures the sync is saved to a local spool and the run exits 12 ("spooled; run `cloudexit spool flush`"). `--spool-on-error` also spools syncs the platform refused, for a later manual retry. The spool is `$XDG_STATE_HOME/cloudexit/spool/<host id>/` (`~/.local/state/...`, or `CLOUDEXIT_SPOOL_DIR`): the sync document and a small file with the attempts and the last error, readable by you only, never holding the API key. The host id is a hash, so the host name is not in the path.
+When the budget runs out on retryable failures the sync is saved to a local spool and the run exits 12 ("spooled; run `cloudexit spool flush`"). `--spool-on-error` also spools syncs the platform refused, for a later manual retry. The spool is `$XDG_STATE_HOME/cloudexit/spool/<host id>/` (`~/.local/state/...`, or `CLOUDEXIT_SPOOL_DIR`): the sync document and a small file with the attempts and the last error, readable by you only, never holding the API key. The host id is a hash, so the host name is not in the path. Each spooled sync also records a fingerprint of the API key in use when it was spooled (a hash, never the key).
 
 ```bash
 cloudexit spool list                  # id, host id, age, attempts, last error
@@ -229,7 +229,9 @@ cloudexit spool purge [--older-than 30d] [--yes]
 cloudexit upload raw_data/sync.json   # upload sync documents (or folders of them), for example from an air-gapped machine
 ```
 
-Every run with a platform first delivers the spooled syncs of the same host (oldest first, within 60 seconds); `--no-flush` skips that, and a failing flush never blocks the new run. A spooled sync older than the platform's `max_collection_age_days` is reported as "will be rejected" and kept until you purge it. For an account key, `upload` needs `--assessment-id`, or `--name` and `--exit-strategy`, to say where the sync goes.
+Every run with a platform first delivers the spooled syncs of the same host and the same API key (oldest first, within 60 seconds); `--no-flush` skips that, and a failing flush never blocks the new run. Syncs spooled with another API key are never delivered with this one: they stay in the spool, the run says so, and `cloudexit spool list` marks them (`cloudexit spool purge` removes them). A spooled sync with no recorded key is treated the same way.
+
+Running again after exit 12 does two things: it delivers the spooled sync under its own sync ID, which the platform stores and charges once however often it is sent, and it then makes a new assessment, which is a new sync with a new sync ID and is charged like any other. Use `--no-flush` to leave the spool alone, or `cloudexit spool flush` to deliver it without a new assessment. A spooled sync older than the platform's `max_collection_age_days` is reported as "will be rejected" and kept until you purge it. For an account key, `upload` needs `--assessment-id`, or `--name` and `--exit-strategy`, to say where the sync goes.
 
 Connections always verify the platform's certificate. For a private certificate authority use `--ca-bundle` or `CLOUDEXIT_CA_BUNDLE`; otherwise `REQUESTS_CA_BUNDLE` and `SSL_CERT_FILE` are honoured. There is no option to turn verification off. `HTTPS_PROXY` and `NO_PROXY` are honoured. Requests give up after 10 seconds to connect and 60 seconds to read (`--timeout` sets both). The platform's capabilities are cached for five minutes in `~/.cache/cloudexit/` (`$XDG_CACHE_HOME/cloudexit/`, or `%LOCALAPPDATA%\cloudexit\cache\` on Windows).
 

cloudexit/catalogue/db.py

@@ -39,6 +39,8 @@ def connect(db_path=None):
     db_path = db_path or active_catalogue_path()
     try:
         conn = sqlite3.connect(db_path, factory=_Connection)
+        # Nothing the file's own SQL defines is trusted to run
+        conn.execute("PRAGMA trusted_schema=OFF")
         return conn
     except sqlite3.Error as e:
         logger.debug(f"Error connecting to database: {e}")

cloudexit/catalogue/download.py

@@ -18,7 +18,6 @@
     check_checksum,
     install_dataset,
     installed_catalogue,
-    migrate_legacy_dataset,
 )
 
 logger = logging.getLogger(__name__)
@@ -203,7 +202,6 @@ def ensure_catalogue(
     base_url: str, *, directory: Path | None = None, verify: str | bool = True
 ) -> EnsureResult:
     directory = directory or catalogue_dir()
-    migrated = migrate_legacy_dataset(directory)
     try:
         result = update_catalogue(base_url, directory=directory, verify=verify)
     except (CatalogueError, OSError) as exc:
@@ -218,7 +216,6 @@ def ensure_catalogue(
             installed.version,
             installed.date,
             str(exc),
-            migrated,
         )
 
     installed = installed_catalogue(directory)
@@ -228,7 +225,6 @@ def ensure_catalogue(
         installed.version,
         installed.date,
         None,
-        migrated,
     )
 
 

cloudexit/catalogue/store.py

@@ -21,7 +21,6 @@
 CATALOGUE_BASE_URL = "https://cloudexit-oss-data-eu.fsn1.your-objectstorage.com"
 SUPPORTED_SCHEMA = 1
 REQUIRED_TABLES = ("resourcetype", "risk")
-LEGACY_DATABASE = Path("datasets") / "data.db"
 PROVIDER_IDS = {"azure": 1, "aws": 2}
 GZIP_MAGIC = b"\x1f\x8b"
 # A dataset unpacks to a few times its download size. The limit is that ratio
@@ -84,7 +83,6 @@ class EnsureResult:
     version: str | None
     date: str
     reason: str | None = None
-    migrated: bool = False
 
 
 def catalogue_dir(
@@ -126,14 +124,45 @@ def catalogue_base_url(settings: Settings) -> str:
 
 
 def _open_readonly(path: Path) -> sqlite3.Connection:
-    return sqlite3.connect(f"{Path(path).resolve().as_uri()}?mode=ro", uri=True)
+    conn = sqlite3.connect(f"{Path(path).resolve().as_uri()}?mode=ro", uri=True)
+    # Nothing the file's own SQL defines is trusted to run
+    conn.execute("PRAGMA trusted_schema=OFF")
+    return conn
 
 
 def _table_names(conn: sqlite3.Connection) -> set[str]:
     rows = conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")
     return {row[0] for row in rows}
 
 
+def _check_schema_objects(conn: sqlite3.Connection) -> None:
+    # A catalogue holds tables and indexes. A trigger or a view runs SQL of its
+    # own on the data a run adds to the same database, so none is accepted.
+    found = conn.execute(
+        "SELECT type, name FROM sqlite_master "
+        "WHERE type NOT IN ('table', 'index') ORDER BY type, name"
+    ).fetchall()
+    if found:
+        names = ", ".join(f"{kind} {name}" for kind, name in found[:3])
+        more = f" and {len(found) - 3} more" if len(found) > 3 else ""
+        raise CatalogueError(
+            f"the catalogue holds {names}{more}, which a catalogue does not have; "
+            "it was not used"
+        )
+
+
+def validate_schema(path: str | Path) -> None:
+    # For a database that is about to be written to, or that has been
+    try:
+        conn = _open_readonly(Path(path))
+        try:
+            _check_schema_objects(conn)
+        finally:
+            conn.close()
+    except sqlite3.DatabaseError as exc:
+        raise CatalogueError(f"not a valid catalogue database: {exc}") from exc
+
+
 def _read_meta(conn: sqlite3.Connection, tables: set[str]) -> CatalogueInfo | None:
     if "catalogue_meta" not in tables:
         return None
@@ -215,6 +244,7 @@ def validate_database(
                 raise CatalogueError(
                     f"not a catalogue database: missing table(s) {', '.join(missing)}"
                 )
+            _check_schema_objects(conn)
             info = _read_meta(conn, tables)
         finally:
             conn.close()
@@ -257,32 +287,6 @@ def installed_catalogue(directory: Path | None = None) -> Installed | None:
     return Installed(path, info, read_state(directory))
 
 
-def migrate_legacy_dataset(
-    directory: Path | None = None, legacy: Path | None = None
-) -> bool:
-    directory = directory or catalogue_dir()
-    legacy = legacy or LEGACY_DATABASE
-    if installed_catalogue(directory) or not legacy.is_file():
-        return False
-    try:
-        info = validate_database(legacy)
-    except CatalogueError:
-        return False
-
-    directory.mkdir(parents=True, exist_ok=True)
-    shutil.copy2(legacy, directory / "data.db")
-    modified = datetime.fromtimestamp(legacy.stat().st_mtime, timezone.utc)
-    _write_state(
-        directory,
-        {
-            "version": info.version if info else None,
-            "sha256": None,
-            "downloaded_at": modified.isoformat(timespec="seconds"),
-        },
-    )
-    return True
-
-
 def _is_gzip(data: bytes) -> bool:
     return data[:2] == GZIP_MAGIC