Security Boundaries & Access Controls in Spatial SQLite

# Python 3.9+ — OS permission verification before opening a spatial database
import os
import stat

def assert_db_permissions(db_path: str, writable: bool = False) -> None:
    """Raise PermissionError if the process cannot perform the requested access.

    os.access() checks effective uid/gid and honours ACLs and read-only
    mounts — it is more reliable than inspecting raw permission bits.
    """
    if not os.path.exists(db_path):
        raise FileNotFoundError(f"Database not found: {db_path}")
    if not os.access(db_path, os.R_OK):
        raise PermissionError(f"No read permission on {db_path}")
    if writable and not os.access(db_path, os.W_OK):
        raise PermissionError(f"No write permission on {db_path}; "
                              "check mount flags or ACLs")

    # Warn when world-readable: geometry and attribute data are exposed to all users
    mode = stat.S_IMODE(os.stat(db_path).st_mode)
    if mode & stat.S_IROTH:
        import warnings
        warnings.warn(
            f"{db_path} is world-readable (mode {oct(mode)}). "
            "Consider chmod o-r for sensitive spatial data.",
            stacklevel=2,
        )

# Mount /dev/sdb1 as read-only; SQLite will honour the OS flag
mount -o ro /dev/sdb1 /mnt/field-data

icacls "C:\GISData\survey.gpkg" /deny "GISService:(W,M)"

# Python 3.9+ — read-only and read-write connection factory for GeoPackage / SpatiaLite
import sqlite3
import logging
from contextlib import contextmanager

logger = logging.getLogger(__name__)

@contextmanager
def spatial_db_connection(db_path: str, read_only: bool = True):
    """Open a GeoPackage or SpatiaLite database with an explicit access mode.

    Yields a sqlite3.Connection; commits on clean exit, rolls back on error.
    WAL mode is only activated on writable connections — switching journal mode
    requires writing the database header, which mode=ro forbids.
    """
    mode = "ro" if read_only else "rwc"
    uri = f"file:{db_path}?mode={mode}"
    conn = None
    try:
        conn = sqlite3.connect(uri, uri=True)
        conn.row_factory = sqlite3.Row
        # Foreign-key enforcement is off by default; enable before any DML
        conn.execute("PRAGMA foreign_keys = ON;")
        if not read_only:
            # WAL mode improves concurrent reader throughput; must be set by a writer
            conn.execute("PRAGMA journal_mode = WAL;")
            conn.execute("PRAGMA synchronous = NORMAL;")
        yield conn
        if not read_only:
            conn.commit()
    except sqlite3.OperationalError:
        logger.exception("SQLite operation failed on %s", db_path)
        if conn:
            conn.rollback()
        raise
    finally:
        if conn:
            conn.close()

# Python 3.9+ + pysqlcipher3 — open an AES-256-encrypted GeoPackage
from pysqlcipher3 import dbapi2 as sqlcipher
import os

def open_encrypted_gpkg(db_path: str) -> sqlcipher.Connection:
    """Open a SQLCipher-encrypted GeoPackage using a key from the environment.

    Never embed the passphrase in source control. Use an environment variable,
    platform keychain, or HSM-backed secret store.
    """
    key = os.environ.get("GPKG_KEY")
    if not key:
        raise RuntimeError("GPKG_KEY environment variable is not set")

    conn = sqlcipher.connect(db_path)
    # Cipher parameters must be set before the first table access
    conn.execute("PRAGMA cipher_compatibility = 4;")
    conn.execute("PRAGMA cipher_page_size = 4096;")
    # PRAGMA cannot use bound parameters; escape single quotes to prevent injection
    safe_key = key.replace("'", "''")
    conn.execute(f"PRAGMA key = '{safe_key}';")

    # Probe sqlite_master: if decryption fails this raises OperationalError
    conn.execute("SELECT count(*) FROM sqlite_master;")
    return conn

# Python 3.9+ + pysqlcipher3 — encrypt an existing GeoPackage in place
import os
from pysqlcipher3 import dbapi2 as sqlcipher

def encrypt_existing_gpkg(src_path: str, dst_path: str) -> None:
    """Copy src_path into a new AES-256-encrypted database at dst_path."""
    key = os.environ["GPKG_KEY"].replace("'", "''")

    src = sqlcipher.connect(src_path)
    src.execute("PRAGMA key = '';")  # open source without encryption

    src.execute(f"ATTACH DATABASE '{dst_path}' AS encrypted KEY '{key}';")
    src.execute("SELECT sqlcipher_export('encrypted');")
    src.execute("DETACH DATABASE encrypted;")
    src.close()

# Python 3.9+ — validate spatial metadata before allowing write access
import sqlite3

def validate_spatial_metadata(
    conn: sqlite3.Connection,
    expected_srid: int = 4326,
) -> None:
    """Raise ValueError if the database lacks conformant spatial metadata.

    Handles both GeoPackage (gpkg_geometry_columns.srs_id) and SpatiaLite
    (geometry_columns.srid), which use different table and column names.
    """
    is_gpkg = conn.execute(
        "SELECT 1 FROM sqlite_master "
        "WHERE type = 'table' AND name = 'gpkg_geometry_columns';"
    ).fetchone()

    if is_gpkg:
        count = conn.execute(
            "SELECT count(*) FROM gpkg_geometry_columns WHERE srs_id = ?;",
            (expected_srid,),
        ).fetchone()[0]
        table_label = "gpkg_geometry_columns"
        col_label = "srs_id"
    else:
        count = conn.execute(
            "SELECT count(*) FROM geometry_columns WHERE srid = ?;",
            (expected_srid,),
        ).fetchone()[0]
        table_label = "geometry_columns"
        col_label = "srid"

    if count == 0:
        raise ValueError(
            f"No geometry tables registered with {col_label}={expected_srid} "
            f"in {table_label}. Verify CRS registration before writing."
        )

# Python 3.9+ — retry decorator for SQLite SQLITE_BUSY / "database is locked"
import time
import random
import sqlite3
from functools import wraps

def retry_on_busy(max_retries: int = 6, base_delay_s: float = 0.05):
    """Retry a function on sqlite3.OperationalError 'database is locked'.

    Uses truncated binary-exponential backoff with jitter to avoid thundering
    herd when multiple workers share a single GeoPackage writer.
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except sqlite3.OperationalError as exc:
                    if "database is locked" not in str(exc).lower():
                        raise
                    if attempt == max_retries - 1:
                        raise
                    delay = min(base_delay_s * (2 ** attempt), 2.0)
                    delay += random.uniform(0, delay * 0.1)
                    time.sleep(delay)
        return wrapper
    return decorator

@retry_on_busy()
def insert_feature(conn: sqlite3.Connection, table: str, geom_wkb: bytes,
                   attributes: dict) -> None:
    """Insert a single feature inside an IMMEDIATE transaction."""
    cols = ", ".join(["geom"] + list(attributes.keys()))
    placeholders = ", ".join(["?"] * (1 + len(attributes)))
    values = [geom_wkb] + list(attributes.values())
    with conn:
        conn.execute("BEGIN IMMEDIATE;")
        conn.execute(
            f"INSERT INTO {table} ({cols}) VALUES ({placeholders});",
            values,
        )

# 1. Confirm the file is not world-readable
stat -c "%a %n" /path/to/survey.gpkg
# Expected: 640 or 600 — not 644 or 777

# 2. Confirm read-only mount prevents writes
sqlite3 /mnt/field-data/survey.gpkg "INSERT INTO gpkg_contents VALUES (null);"
# Expected: Error: attempt to write a readonly database

# 3. Confirm encrypted file cannot be read without the key
python3 -c "
import sqlite3, sys
try:
    c = sqlite3.connect(sys.argv[1])
    c.execute('SELECT count(*) FROM sqlite_master').fetchone()
    print('WARNING: file is unencrypted or key matched empty string')
except sqlite3.DatabaseError:
    print('OK: file is encrypted / inaccessible without SQLCipher')
" /path/to/encrypted.gpkg

# 4. Confirm WAL mode is active (only meaningful on writable connections)
sqlite3 /path/to/writable.gpkg "PRAGMA journal_mode;"
# Expected: wal

# Python 3.9+ — verify that a read-only connection rejects writes
import sqlite3

def assert_read_only(db_path: str) -> None:
    uri = f"file:{db_path}?mode=ro"
    conn = sqlite3.connect(uri, uri=True)
    try:
        conn.execute("CREATE TABLE _rw_test (id INTEGER PRIMARY KEY);")
        raise AssertionError("Connection accepted a DDL write — mode=ro not enforced")
    except sqlite3.OperationalError as exc:
        assert "readonly" in str(exc).lower(), f"Unexpected error: {exc}"
    finally:
        conn.close()

# GeoPackage — enable WAL before making the file read-only
with spatial_db_connection(db_path, read_only=False) as conn:
    result = conn.execute("PRAGMA journal_mode = WAL;").fetchone()[0]
    assert result == "wal", f"WAL not enabled: got {result}"
# Now safe to chmod or remount read-only

# Diagnose by probing compatibility versions
for compat in (3, 4):
    try:
        c = sqlcipher.connect(db_path)
        c.execute(f"PRAGMA cipher_compatibility = {compat};")
        c.execute(f"PRAGMA key = '{safe_key}';")
        c.execute("SELECT count(*) FROM sqlite_master;")
        print(f"Opened successfully with cipher_compatibility = {compat}")
        c.close()
        break
    except Exception as exc:
        print(f"compat={compat} failed: {exc}")

-- GeoPackage SQL — register a feature table that was inserted without metadata
INSERT OR REPLACE INTO gpkg_contents (
    table_name, data_type, identifier, description, last_change, srs_id
) VALUES ('my_features', 'features', 'my_features', '', datetime('now'), 4326);

INSERT OR REPLACE INTO gpkg_geometry_columns (
    table_name, column_name, geometry_type_name, srs_id, z, m
) VALUES ('my_features', 'geom', 'GEOMETRY', 4326, 0, 0);

# Python 3.9+ — configure busy_timeout for multi-process pipelines
conn = sqlite3.connect(db_path)
conn.execute("PRAGMA busy_timeout = 5000;")  # wait up to 5 s before raising
conn.execute("BEGIN IMMEDIATE;")
# ... DML ...
conn.commit()

import os, shutil, stat

def secure_copy_gpkg(src: str, dst: str, mode: int = 0o640) -> None:
    shutil.copy2(src, dst)
    os.chmod(dst, mode)
    # Verify
    actual = stat.S_IMODE(os.stat(dst).st_mode)
    assert actual == mode, f"chmod failed: got {oct(actual)}"