From 688262ab92ec4a91c03a604f092c912a64fd36c4 Mon Sep 17 00:00:00 2001 From: Peter Adams <18162810+Maxteabag@users.noreply.github.com> Date: Tue, 14 Jul 2026 20:11:43 +0200 Subject: [PATCH] Let Oracle users express secure Easy Connect DSNs Oracle service connections need optional TCP/TCPS protocol selection and free-form Easy Connect parameters. Keep host, port, and service structured so existing form generation and SSH endpoint rewriting continue to work, then compose the supported values into the driver DSN. Constraint: Preserve existing host, port, service-name, SID, and SSH behavior Rejected: Add a raw DSN override | bypasses structured endpoint and SSH tunnel rewriting Confidence: high Scope-risk: narrow Directive: Easy Connect options apply only to service-name connections; keep SID descriptors on oracledb.makedsn Tested: 1051 unit tests; 300 UI tests; 17 focused Oracle tests; python-oracledb ConnectParams parsing; targeted Ruff Not-tested: Live TCPS connection to an Oracle listener Related: #261 --- .../connections/providers/oracle/adapter.py | 13 +- .../connections/providers/oracle/schema.py | 23 ++++ .../providers/oracle_legacy/schema.py | 23 ++++ .../test_oracle_easy_connect_cli_config.py | 42 ++++++ tests/unit/test_oracle_adapter.py | 126 ++++++++++++++++++ 5 files changed, 226 insertions(+), 1 deletion(-) create mode 100644 tests/integration/test_oracle_easy_connect_cli_config.py diff --git a/sqlit/domains/connections/providers/oracle/adapter.py b/sqlit/domains/connections/providers/oracle/adapter.py index 76edc895..a79f5b27 100644 --- a/sqlit/domains/connections/providers/oracle/adapter.py +++ b/sqlit/domains/connections/providers/oracle/adapter.py @@ -83,7 +83,18 @@ def connect(self, config: ConnectionConfig) -> Any: sid = config.get_option("oracle_sid") or endpoint.database dsn = oracledb.makedsn(endpoint.host, port, sid=sid) else: - dsn = f"{endpoint.host}:{port}/{endpoint.database}" + protocol = str(config.get_option("oracle_protocol", "default")).strip().lower() + if protocol not in {"", "default", "tcp", "tcps"}: + raise ValueError("Oracle protocol must be Default, TCP, or TCPS") + protocol_prefix = f"{protocol}://" if protocol in {"tcp", "tcps"} else "" + dsn = f"{protocol_prefix}{endpoint.host}:{port}/{endpoint.database}" + + parameters = str( + config.get_option("oracle_easy_connect_parameters", "") or "" + ).strip() + parameters = parameters.lstrip("?") + if parameters: + dsn = f"{dsn}?{parameters}" # Determine connection mode based on oracle_role oracle_role = config.get_option("oracle_role", "normal") diff --git a/sqlit/domains/connections/providers/oracle/schema.py b/sqlit/domains/connections/providers/oracle/schema.py index f004a6d9..7aee63c9 100644 --- a/sqlit/domains/connections/providers/oracle/schema.py +++ b/sqlit/domains/connections/providers/oracle/schema.py @@ -27,6 +27,14 @@ def _get_oracle_connection_type_options() -> tuple[SelectOption, ...]: ) +def _get_oracle_protocol_options() -> tuple[SelectOption, ...]: + return ( + SelectOption("default", "Default"), + SelectOption("tcp", "TCP"), + SelectOption("tcps", "TCPS"), + ) + + def _oracle_connection_type_is_service_name(values: dict) -> bool: return values.get("oracle_connection_type", "service_name") != "sid" @@ -61,6 +69,21 @@ def _oracle_connection_type_is_sid(values: dict) -> bool: required=True, visible_when=_oracle_connection_type_is_service_name, ), + SchemaField( + name="oracle_protocol", + label="Protocol", + field_type=FieldType.DROPDOWN, + options=_get_oracle_protocol_options(), + default="default", + visible_when=_oracle_connection_type_is_service_name, + ), + SchemaField( + name="oracle_easy_connect_parameters", + label="Easy Connect Parameters", + placeholder="ssl_server_dn_match=no&retry_count=3", + description="Oracle Easy Connect parameters, without the leading question mark.", + visible_when=_oracle_connection_type_is_service_name, + ), SchemaField( name="oracle_sid", label="SID", diff --git a/sqlit/domains/connections/providers/oracle_legacy/schema.py b/sqlit/domains/connections/providers/oracle_legacy/schema.py index 4252f721..1cac2353 100644 --- a/sqlit/domains/connections/providers/oracle_legacy/schema.py +++ b/sqlit/domains/connections/providers/oracle_legacy/schema.py @@ -27,6 +27,14 @@ def _get_oracle_connection_type_options() -> tuple[SelectOption, ...]: ) +def _get_oracle_protocol_options() -> tuple[SelectOption, ...]: + return ( + SelectOption("default", "Default"), + SelectOption("tcp", "TCP"), + SelectOption("tcps", "TCPS"), + ) + + def _oracle_connection_type_is_service_name(values: dict) -> bool: return values.get("oracle_connection_type", "service_name") != "sid" @@ -72,6 +80,21 @@ def _oracle_thick_mode_enabled(values: dict) -> bool: required=True, visible_when=_oracle_connection_type_is_service_name, ), + SchemaField( + name="oracle_protocol", + label="Protocol", + field_type=FieldType.DROPDOWN, + options=_get_oracle_protocol_options(), + default="default", + visible_when=_oracle_connection_type_is_service_name, + ), + SchemaField( + name="oracle_easy_connect_parameters", + label="Easy Connect Parameters", + placeholder="ssl_server_dn_match=no&retry_count=3", + description="Oracle Easy Connect parameters, without the leading question mark.", + visible_when=_oracle_connection_type_is_service_name, + ), SchemaField( name="oracle_sid", label="SID", diff --git a/tests/integration/test_oracle_easy_connect_cli_config.py b/tests/integration/test_oracle_easy_connect_cli_config.py new file mode 100644 index 00000000..c9e74faf --- /dev/null +++ b/tests/integration/test_oracle_easy_connect_cli_config.py @@ -0,0 +1,42 @@ +"""Regression coverage for Oracle Easy Connect CLI configuration.""" + +from __future__ import annotations + +import argparse + +import pytest + +from sqlit.domains.connections.cli.helpers import ( + add_schema_arguments, + build_connection_config_from_args, +) +from sqlit.domains.connections.providers.catalog import get_provider_schema + + +@pytest.mark.parametrize("db_type", ["oracle", "oracle_legacy"]) +def test_cli_builds_oracle_easy_connect_options_into_config(db_type: str) -> None: + schema = get_provider_schema(db_type) + parser = argparse.ArgumentParser() + add_schema_arguments(parser, schema, include_name=True, name_required=True) + + args = parser.parse_args( + [ + "--name", + "secure-oracle", + "--server", + "localhost", + "--database", + "service-name.com", + "--username", + "testuser", + "--oracle-protocol", + "tcps", + "--oracle-easy-connect-parameters", + "ssl_server_dn_match=no&retry_count=3", + ] + ) + + config = build_connection_config_from_args(schema, args, name=args.name) + + assert config.get_option("oracle_protocol") == "tcps" + assert config.get_option("oracle_easy_connect_parameters") == ("ssl_server_dn_match=no&retry_count=3") diff --git a/tests/unit/test_oracle_adapter.py b/tests/unit/test_oracle_adapter.py index 7693b76e..5c02a36e 100644 --- a/tests/unit/test_oracle_adapter.py +++ b/tests/unit/test_oracle_adapter.py @@ -161,6 +161,132 @@ def test_connect_service_name_format(self): # Service name uses slash separator: host:port/service_name assert call_kwargs["dsn"] == "localhost:1521/XEPDB1" + def test_connect_tcps_with_easy_connect_parameters(self): + """Issue #261: protocol and parameters must be included in the DSN.""" + mock_oracledb = MagicMock() + mock_oracledb.AUTH_MODE_SYSDBA = 2 + mock_oracledb.AUTH_MODE_SYSOPER = 4 + + with patch.dict("sys.modules", {"oracledb": mock_oracledb}): + from sqlit.domains.connections.providers.oracle.adapter import OracleAdapter + + adapter = OracleAdapter() + config = ConnectionConfig( + name="test", + db_type="oracle", + server="localhost", + port="1521", + database="service-name.com", + username="testuser", + password="testpass", + options={ + "oracle_connection_type": "service_name", + "oracle_protocol": "tcps", + "oracle_easy_connect_parameters": ( + "ssl_server_dn_match=no&retry_count=3" + ), + }, + ) + + adapter.connect(config) + + call_kwargs = mock_oracledb.connect.call_args.kwargs + assert call_kwargs["dsn"] == ( + "tcps://localhost:1521/service-name.com" + "?ssl_server_dn_match=no&retry_count=3" + ) + + def test_connect_easy_connect_parameters_accept_leading_question_mark(self): + """Users may paste Easy Connect parameters with their separator.""" + mock_oracledb = MagicMock() + + with patch.dict("sys.modules", {"oracledb": mock_oracledb}): + from sqlit.domains.connections.providers.oracle.adapter import OracleAdapter + + config = ConnectionConfig( + name="test", + db_type="oracle", + server="localhost", + port="1521", + database="XEPDB1", + username="testuser", + password="testpass", + options={"oracle_easy_connect_parameters": "?expire_time=2"}, + ) + + OracleAdapter().connect(config) + + assert mock_oracledb.connect.call_args.kwargs["dsn"] == ( + "localhost:1521/XEPDB1?expire_time=2" + ) + + def test_connect_rejects_unknown_oracle_protocol(self): + """Configs outside the schema must not silently discard bad protocols.""" + mock_oracledb = MagicMock() + + with patch.dict("sys.modules", {"oracledb": mock_oracledb}): + from sqlit.domains.connections.providers.oracle.adapter import OracleAdapter + + config = ConnectionConfig( + name="test", + db_type="oracle", + server="localhost", + port="1521", + database="XEPDB1", + username="testuser", + password="testpass", + options={"oracle_protocol": "udp"}, + ) + + with pytest.raises(ValueError, match="Default, TCP, or TCPS"): + OracleAdapter().connect(config) + + mock_oracledb.connect.assert_not_called() + + def test_connect_sid_ignores_easy_connect_options(self): + """Easy Connect protocol and parameters do not apply to SID descriptors.""" + mock_oracledb = MagicMock() + + with patch.dict("sys.modules", {"oracledb": mock_oracledb}): + from sqlit.domains.connections.providers.oracle.adapter import OracleAdapter + + config = ConnectionConfig( + name="test", + db_type="oracle", + server="localhost", + port="1521", + username="testuser", + password="testpass", + options={ + "oracle_connection_type": "sid", + "oracle_sid": "ORCL", + "oracle_protocol": "tcps", + "oracle_easy_connect_parameters": "ssl_server_dn_match=no", + }, + ) + + OracleAdapter().connect(config) + + mock_oracledb.makedsn.assert_called_once_with("localhost", 1521, sid="ORCL") + assert mock_oracledb.connect.call_args.kwargs["dsn"] is ( + mock_oracledb.makedsn.return_value + ) + + def test_tcps_easy_connect_dsn_parses_in_real_driver(self): + """The issue #261 DSN must be accepted by python-oracledb itself.""" + oracledb = pytest.importorskip("oracledb") + params = oracledb.ConnectParams() + + params.parse_connect_string( + "tcps://localhost:1521/service-name.com?ssl_server_dn_match=no" + ) + + assert params.protocol == "tcps" + assert params.host == "localhost" + assert params.port == 1521 + assert params.service_name == "service-name.com" + assert params.ssl_server_dn_match is False + def test_connect_sid_format(self): """SID connection type must go through oracledb.makedsn — see issue #106.