fix: enforce read-only connection mode and report the actual catalog

docs/test-results/2026-11-02-access-mode-and-catalog.md

@@ -0,0 +1,30 @@
+# Connection access mode and catalog regression
+
+Validated on 2026-11-02 with a native macOS ARM64 build, CPython 3.12 and a real DM8 database. No database proxy or Linux container was used for the client.
+
+## Reproduced behavior
+
+The original bridge accepted `access_mode=dmPython.DSQL_MODE_READ_ONLY` but reported mode 0 and allowed a zero-row DELETE. It also returned an empty `current_catalog`, although `SELECT NAME FROM V$DATABASE` returned a database name. All seven new access-mode/catalog regression cases failed against the original bridge.
+
+## Fixed behavior
+
+- Apply read-only mode on the server before returning a successful login, and report the effective requested mode.
+- Keep the read-only session pinned in autocommit mode. Reuse it when disabling autocommit, and reapply protection after enabling autocommit. Close the connection if protection cannot be restored.
+- Apply runtime mode changes and invalidate prepared statements when their connection binding changes. Reject invalid modes without relaxing the current mode.
+- Read the current catalog from the connected database, propagating metadata query errors.
+
+The following command passed **20 tests with no skips**, including all seven new regressions:
+
+```bash
+python -m pytest \
+  tests/integration/test_p1_connection_access_mode.py \
+  tests/integration/test_p1_contract.py \
+  tests/integration/test_p1_cursor_state_contract.py \
+  --require-dm -p no:randomly -q
+```
+
+`DM_TEST_HOST`, `DM_TEST_PORT`, `DM_TEST_USER` and `DM_TEST_PASSWORD` must point to a dedicated test schema. The new tests create and remove their own randomly named table. Their write-rejection checks use zero-row DELETE statements.
+
+The bridge was also exercised through the unmodified official 2.5.38 C extension in a downstream application: connection testing, catalog/schema discovery, parameterized read-only queries, eight concurrent queries, write rejection and recovery after invalid credentials passed. Source-table snapshots before and after that application check were identical.
+
+This does not establish complete interchangeability with the vendor DPI SDK. An additional `executemany` error-contract check failed with the official 2.5.38 C extension against both the original and patched bridge; the same check passed with this fork's C extension and is included in the 20-test result above. No vendor SDK or compiled driver artifact is distributed by this change.

dpi_bridge/dpi_conn.go

@@ -42,6 +42,11 @@ import (
 	dm "gitee.com/chunanyong/dm"
 )
 
+const (
+	accessModeReadWrite = 0
+	accessModeReadOnly  = 1
+)
+
 // connHandle represents a DPI connection.
 type connHandle struct {
 	mu             sync.Mutex
@@ -59,6 +64,7 @@ type connHandle struct {
 	password      string
 	schema        string
 	autocommit    bool
+	accessMode    int
 	loginTimeout  int
 	connTimeout   int
 	appName       string
@@ -162,16 +168,36 @@ func dpi_set_con_attr(hcon C.dhcon, attrID C.sdint4, val C.dpointer, valLen C.sd
 					conn.lastErr = diagFromError(dbErr)
 					return DSQL_ERROR
 				}
-				conn.pinned.Close()
-				conn.pinned = nil
+				if conn.accessMode == accessModeReadOnly {
+					if err := setSessionAccessMode(conn.pinned, conn.accessMode); err != nil {
+						// Changing autocommit can reset the driver's read-only
+						// transaction flag. Never keep an unprotected session.
+						conn.pinned.Close()
+						conn.pinned = nil
+						conn.db.Close()
+						conn.db = nil
+						conn.conn = nil
+						conn.lastErr = diagFromError(err)
+						return DSQL_ERROR
+					}
+				} else {
+					conn.pinned.Close()
+					conn.pinned = nil
+				}
 			} else {
-				pinned, dbErr := conn.db.Conn(context.Background())
-				if dbErr != nil {
-					conn.lastErr = diagFromError(dbErr)
-					return DSQL_ERROR
+				pinned := conn.pinned
+				if pinned == nil {
+					var dbErr error
+					pinned, dbErr = conn.db.Conn(context.Background())
+					if dbErr != nil {
+						conn.lastErr = diagFromError(dbErr)
+						return DSQL_ERROR
+					}
 				}
-				if dbErr = setPinnedAutoCommit(pinned, false); dbErr != nil {
-					pinned.Close()
+				if dbErr := setPinnedAutoCommit(pinned, false); dbErr != nil {
+					if pinned != conn.pinned {
+						pinned.Close()
+					}
 					conn.lastErr = diagFromError(dbErr)
 					return DSQL_ERROR
 				}
@@ -182,6 +208,11 @@ func dpi_set_con_attr(hcon C.dhcon, attrID C.sdint4, val C.dpointer, valLen C.sd
 		if changed {
 			conn.modeGeneration++
 		}
+	case DSQL_ATTR_ACCESS_MODE:
+		if err := conn.setAccessMode(intVal); err != nil {
+			conn.lastErr = diagFromError(err)
+			return DSQL_ERROR
+		}
 	case DSQL_ATTR_LOGIN_TIMEOUT:
 		if intVal < 0 {
 			conn.lastErr = &diagInfo{errorCode: -1, message: "login timeout must be non-negative"}
@@ -311,7 +342,7 @@ func dpi_set_con_attr(hcon C.dhcon, attrID C.sdint4, val C.dpointer, valLen C.sd
 		DSQL_ATTR_OSAUTH_TYPE, DSQL_ATTR_DDL_AUTOCOMMIT,
 		DSQL_ATTR_COMPATIBLE_MODE, DSQL_ATTR_SHAKE_CRYPTO,
 		DSQL_ATTR_NLS_NUMERIC_CHARACTERS,
-		DSQL_ATTR_ACCESS_MODE, DSQL_ATTR_PACKET_SIZE,
+		DSQL_ATTR_PACKET_SIZE,
 		DSQL_ATTR_CURRENT_CATALOG:
 		// Accept but ignore these for now
 	default:
@@ -330,6 +361,48 @@ func setPinnedAutoCommit(pinned *sql.Conn, enabled bool) error {
 	})
 }
 
+func setSessionAccessMode(pinned *sql.Conn, mode int) error {
+	_, err := pinned.ExecContext(context.Background(), fmt.Sprintf("SP_SET_SESSION_READONLY(%d)", mode))
+	return err
+}
+
+// Caller holds conn.mu. Read-only sessions remain pinned even in autocommit
+// mode, so a pooled/reconnected session cannot silently lose this setting.
+func (conn *connHandle) setAccessMode(mode int) error {
+	if mode != accessModeReadWrite && mode != accessModeReadOnly {
+		return fmt.Errorf("access_mode must be READ_WRITE (0) or READ_ONLY (1)")
+	}
+	if conn.accessMode == mode {
+		return nil
+	}
+	if conn.db == nil {
+		conn.accessMode = mode
+		return nil
+	}
+	pinned := conn.pinned
+	if pinned == nil {
+		var err error
+		pinned, err = conn.db.Conn(context.Background())
+		if err != nil {
+			return err
+		}
+	}
+	if err := setSessionAccessMode(pinned, mode); err != nil {
+		if pinned != conn.pinned {
+			pinned.Close()
+		}
+		return err
+	}
+	conn.accessMode = mode
+	conn.modeGeneration++
+	conn.pinned = pinned
+	if mode == accessModeReadWrite && conn.autocommit {
+		conn.pinned.Close()
+		conn.pinned = nil
+	}
+	return nil
+}
+
 //export dpi_get_con_attr
 func dpi_get_con_attr(hcon C.dhcon, attrID C.sdint4, val C.dpointer, bufLen C.sdint4, valLen *C.sdint4) C.DPIRETURN {
 	conn, err := getConnHandle(hcon)
@@ -407,6 +480,31 @@ func dpi_get_con_attr(hcon C.dhcon, attrID C.sdint4, val C.dpointer, bufLen C.sd
 		if valLen != nil {
 			*valLen = 4
 		}
+	case DSQL_ATTR_ACCESS_MODE:
+		*(*C.udint4)(val) = C.udint4(conn.accessMode)
+		if valLen != nil {
+			*valLen = 4
+		}
+	case DSQL_ATTR_CURRENT_CATALOG:
+		if conn.db == nil {
+			conn.lastErr = &diagInfo{errorCode: -1, message: "Not connected"}
+			return DSQL_ERROR
+		}
+		var row *sql.Row
+		if conn.pinned != nil {
+			row = conn.pinned.QueryRowContext(context.Background(), "SELECT NAME FROM V$DATABASE")
+		} else {
+			row = conn.db.QueryRow("SELECT NAME FROM V$DATABASE")
+		}
+		var catalog string
+		if err := row.Scan(&catalog); err != nil {
+			conn.lastErr = diagFromError(err)
+			return DSQL_ERROR
+		}
+		n := cStringLen((*C.sdbyte)(val), int(bufLen), catalog)
+		if valLen != nil {
+			*valLen = C.sdint4(n)
+		}
 	case DSQL_ATTR_TXN_ISOLATION:
 		*(*C.sdint4)(val) = C.sdint4(conn.txnIsolation)
 		if valLen != nil {
@@ -631,10 +729,18 @@ func dpi_login(hcon C.dhcon, svr *C.sdbyte, user *C.sdbyte, pwd *C.sdbyte) C.DPI
 		conn.lastErr = &diagInfo{errorCode: -1, message: "ssl_path requested, but the server did not negotiate encrypted SSL"}
 		return DSQL_ERROR
 	}
+	if conn.accessMode == accessModeReadOnly {
+		if err := setSessionAccessMode(rawConn, conn.accessMode); err != nil {
+			rawConn.Close()
+			db.Close()
+			conn.lastErr = diagFromError(err)
+			return DSQL_ERROR
+		}
+	}
 
 	conn.db = db
 	conn.conn = dmConn
-	if !conn.autocommit {
+	if !conn.autocommit || conn.accessMode == accessModeReadOnly {
 		conn.pinned = rawConn
 	}
 	if dmConn.CompressionMode() != 0 {
@@ -658,7 +764,7 @@ func dpi_login(hcon C.dhcon, svr *C.sdbyte, user *C.sdbyte, pwd *C.sdbyte) C.DPI
 			conn.serverCode = PG_UTF8
 		}
 	}
-	if conn.autocommit {
+	if conn.autocommit && conn.accessMode == accessModeReadWrite {
 		rawConn.Close()
 	}
 

tests/integration/test_p1_connection_access_mode.py

@@ -0,0 +1,100 @@
+"""Connection access mode and catalog must reflect the real DM session."""
+
+from __future__ import annotations
+
+import pytest
+
+import dmPython
+
+
+pytestmark = [pytest.mark.requires_dm, pytest.mark.p1_contract]
+
+
+@pytest.fixture()
+def access_mode_table(conn_params, table_name_factory):
+    name = table_name_factory("DMPY_TEST_ACCESS")
+    conn = dmPython.connect(**conn_params, autoCommit=True)
+    try:
+        with conn.cursor() as cur:
+            cur.execute(f"CREATE TABLE {name} (ID INT PRIMARY KEY)")
+            cur.execute(f"INSERT INTO {name} VALUES (1)")
+        yield name
+    finally:
+        try:
+            with conn.cursor() as cur:
+                cur.execute(f"DROP TABLE {name}")
+        finally:
+            conn.close()
+
+
+def assert_read_only(conn, table):
+    assert conn.access_mode == dmPython.DSQL_MODE_READ_ONLY
+    with conn.cursor() as cur:
+        cur.execute(f"SELECT COUNT(*) FROM {table}")
+        assert cur.fetchone() == (1,)
+        # A zero-row write still requires write access, without modifying rows
+        # when this regression is run against the original buggy library.
+        with pytest.raises(dmPython.DatabaseError, match="6506"):
+            cur.execute(f"DELETE FROM {table} WHERE ID = ?", (-1,))
+
+
+@pytest.mark.parametrize("autocommit", [False, True])
+def test_read_only_access_mode_is_enforced(conn_params, access_mode_table, autocommit):
+    with dmPython.connect(
+        **conn_params,
+        autoCommit=autocommit,
+        access_mode=dmPython.DSQL_MODE_READ_ONLY,
+    ) as conn:
+        assert_read_only(conn, access_mode_table)
+
+
+def test_read_only_survives_transaction_and_autocommit_changes(conn_params, access_mode_table):
+    with dmPython.connect(
+        **conn_params, access_mode=dmPython.DSQL_MODE_READ_ONLY, autoCommit=True
+    ) as conn:
+        for autocommit in [False, True, False, True]:
+            conn.autoCommit = autocommit
+            assert_read_only(conn, access_mode_table)
+            conn.commit()
+            assert_read_only(conn, access_mode_table)
+            conn.rollback()
+            assert_read_only(conn, access_mode_table)
+
+
+@pytest.mark.parametrize("autocommit", [False, True])
+def test_access_mode_changes_rebind_existing_statements(conn_params, access_mode_table, autocommit):
+    with dmPython.connect(**conn_params, autoCommit=autocommit) as conn:
+        with conn.cursor() as cur:
+            sql = f"DELETE FROM {access_mode_table} WHERE ID = ?"
+            cur.execute(sql, (-1,))
+            conn.commit()
+            conn.access_mode = dmPython.DSQL_MODE_READ_ONLY
+            with pytest.raises(dmPython.DatabaseError, match="6506"):
+                cur.execute(sql, (-1,))
+            conn.rollback()
+            conn.access_mode = dmPython.DSQL_MODE_READ_WRITE
+            assert conn.access_mode == dmPython.DSQL_MODE_READ_WRITE
+            cur.execute(sql, (-1,))
+            conn.commit()
+            cur.execute(f"SELECT COUNT(*) FROM {access_mode_table}")
+            assert cur.fetchone() == (1,)
+
+
+def test_invalid_access_mode_does_not_relax_read_only(conn_params, access_mode_table):
+    with dmPython.connect(
+        **conn_params, autoCommit=True, access_mode=dmPython.DSQL_MODE_READ_ONLY
+    ) as conn:
+        with pytest.raises(dmPython.DatabaseError, match="access_mode"):
+            conn.access_mode = 2
+        assert_read_only(conn, access_mode_table)
+
+
+def test_current_catalog_comes_from_the_connected_database(conn_params):
+    with dmPython.connect(
+        **conn_params, autoCommit=True, access_mode=dmPython.DSQL_MODE_READ_ONLY
+    ) as conn:
+        with conn.cursor() as cur:
+            cur.execute("SELECT NAME FROM V$DATABASE")
+            expected = cur.fetchone()[0]
+        assert conn.current_catalog == expected
+        assert expected