fix(trino): harden ticket authentication defaults

README.md

@@ -309,7 +309,7 @@ pipx inject sqlit-tui 'trino[kerberos]'
 python -m pip install 'trino[kerberos]'
 ```
 
-Obtain a valid ticket, for example with `kinit`, then create or edit a Trino connection and choose **Kerberos** as its authentication method. Set the service name and hostname override only when your server's Kerberos principal requires them. Mutual authentication defaults to Optional. Select **GSSAPI** instead when your environment requires the `requests-gssapi` implementation, after installing `trino[gssapi]`; GSSAPI requires a hostname override when a service name is set.
+Obtain a valid ticket, for example with `kinit`, then create or edit a Trino connection and choose **Kerberos** as its authentication method. The service name defaults to `HTTP`; set it or the hostname override only when your server's Kerberos principal requires different values. Mutual authentication uses the selected driver's default unless explicitly overridden. Select **GSSAPI** instead when your environment requires the `requests-gssapi` implementation, after installing `trino[gssapi]`; sqlit derives its default service target from `HTTP` and the configured Trino hostname.
 
 ### SSH Tunnel Support
 

sqlit/core/connection_manager.py

@@ -6,6 +6,7 @@
 from typing import Any
 
 from sqlit.domains.connections.domain.config import ConnectionConfig
+from sqlit.domains.connections.domain.passwords import uses_db_password
 from sqlit.shared.app.services import AppServices
 
 
@@ -31,7 +32,7 @@ def populate_credentials(self, config: ConnectionConfig) -> ConnectionConfig:
             return config
 
         service = self._services.credentials_service
-        if endpoint and endpoint.password is None:
+        if endpoint and endpoint.password is None and uses_db_password(config):
             password = service.get_password(config.name)
             if password is not None:
                 endpoint.password = password

sqlit/domains/connections/app/connection_flow.py

@@ -11,7 +11,7 @@
     PasswordCommandError,
     run_password_command,
 )
-from sqlit.domains.connections.domain.passwords import needs_db_password, needs_ssh_password
+from sqlit.domains.connections.domain.passwords import needs_db_password, needs_ssh_password, uses_db_password
 from sqlit.shared.app import AppServices
 
 
@@ -49,7 +49,7 @@ def populate_credentials_if_missing(self, config: ConnectionConfig) -> None:
         if endpoint and endpoint.password is not None and (not config.tunnel or config.tunnel.password is not None):
             return
         service = self.services.credentials_service
-        if endpoint and endpoint.password is None:
+        if endpoint and endpoint.password is None and uses_db_password(config):
             password = service.get_password(config.name)
             if password is not None:
                 endpoint.password = password

sqlit/domains/connections/cli/prompts.py

@@ -10,6 +10,7 @@
     PasswordCommandError,
     run_password_command,
 )
+from sqlit.domains.connections.domain.passwords import uses_db_password
 
 
 def _needs_ssh_prompt(config: ConnectionConfig) -> bool:
@@ -58,7 +59,7 @@ def prompt_for_password(config: ConnectionConfig) -> ConnectionConfig:
 
     # DB password
     endpoint = config.tcp_endpoint
-    if endpoint and endpoint.password is None:
+    if endpoint and endpoint.password is None and uses_db_password(config):
         if endpoint.password_command:
             try:
                 db_password = run_password_command(endpoint.password_command)

sqlit/domains/connections/domain/config.py

@@ -108,6 +108,8 @@ class TcpEndpoint:
     password: str | None = None
     password_command: str | None = None
     kind: str = "tcp"
+    # Original server name when the transport endpoint is rewritten, for example by an SSH tunnel.
+    original_host: str | None = None
 
 
 @dataclass
@@ -369,6 +371,8 @@ def with_endpoint(self, **kwargs: Any) -> ConnectionConfig:
 
         if not isinstance(self.endpoint, TcpEndpoint):
             return self
+        if "host" in kwargs and kwargs["host"] != self.endpoint.host and "original_host" not in kwargs:
+            kwargs["original_host"] = self.endpoint.original_host or self.endpoint.host
         endpoint = replace(self.endpoint, **kwargs)
         return replace(self, endpoint=endpoint)
 

sqlit/domains/connections/domain/passwords.py

@@ -6,8 +6,8 @@
 from sqlit.domains.connections.providers.metadata import is_file_based, requires_auth
 
 
-def needs_db_password(config: ConnectionConfig) -> bool:
-    """Return True if the database password should be prompted."""
+def uses_db_password(config: ConnectionConfig) -> bool:
+    """Return whether the selected database authentication consumes a password."""
     if is_file_based(config.db_type):
         return False
 
@@ -23,6 +23,14 @@ def needs_db_password(config: ConnectionConfig) -> bool:
         if auth_method in {"none", "kerberos", "gssapi"}:
             return False
 
+    return config.tcp_endpoint is not None
+
+
+def needs_db_password(config: ConnectionConfig) -> bool:
+    """Return True if the database password should be prompted."""
+    if not uses_db_password(config):
+        return False
+
     endpoint = config.tcp_endpoint
     if not endpoint or endpoint.password is not None:
         return False

sqlit/domains/connections/providers/trino/adapter.py

@@ -107,14 +107,15 @@ def connect(self, config: ConnectionConfig) -> Any:
         if schema:
             connect_args["schema"] = schema
 
-        auth = self._build_authentication(config, endpoint.username, endpoint.password)
+        auth_hostname = endpoint.original_host or endpoint.host
+        auth = self._build_authentication(config, endpoint.username, endpoint.password, auth_hostname)
         if auth is not None:
             connect_args["auth"] = auth
 
         connect_args.update({key: value for key, value in config.extra_options.items() if key not in self._AUTH_OPTION_NAMES})
         return trino_dbapi.connect(**connect_args)
 
-    def _build_authentication(self, config: ConnectionConfig, username: str, password: str | None) -> Any | None:
+    def _build_authentication(self, config: ConnectionConfig, username: str, password: str | None, hostname: str) -> Any | None:
         default_method = "basic" if password else "none"
         auth_method = str(self._get_authentication_option(config, "trino_auth_method", default_method)).lower()
 
@@ -173,21 +174,22 @@ def _build_authentication(self, config: ConnectionConfig, username: str, passwor
         auth_args: dict[str, Any] = {
             "delegate": str(self._get_authentication_option(config, "trino_kerberos_delegate", "false")).lower() == "true",
         }
-        mutual_authentication = str(self._get_authentication_option(config, "trino_kerberos_mutual_authentication", "optional")).lower()
+        mutual_authentication = str(self._get_authentication_option(config, "trino_kerberos_mutual_authentication", "driver")).lower()
         mutual_authentication_values = {
             "required": REQUIRED,
             "optional": OPTIONAL,
             "disabled": DISABLED,
         }
-        if mutual_authentication not in mutual_authentication_values:
+        if mutual_authentication != "driver" and mutual_authentication not in mutual_authentication_values:
             raise ValueError(f"Unsupported Trino Kerberos mutual authentication mode: {mutual_authentication}")
-        auth_args["mutual_authentication"] = mutual_authentication_values[mutual_authentication]
+        if mutual_authentication != "driver":
+            auth_args["mutual_authentication"] = mutual_authentication_values[mutual_authentication]
         service_name = self._get_authentication_option(config, "trino_kerberos_service_name")
-        if auth_method == "kerberos" and not service_name:
+        if not service_name:
             service_name = "HTTP"
         hostname_override = self._get_authentication_option(config, "trino_kerberos_hostname_override")
-        if auth_method == "gssapi" and service_name and not hostname_override:
-            raise ValueError("Trino GSSAPI authentication requires a hostname override when a service name is set.")
+        if auth_method == "gssapi" and not hostname_override:
+            hostname_override = hostname
         if service_name:
             auth_args["service_name"] = str(service_name)
         if hostname_override:

sqlit/domains/connections/providers/trino/schema.py

@@ -30,6 +30,7 @@ def _get_authentication_options() -> tuple[SelectOption, ...]:
 
 def _get_kerberos_mutual_authentication_options() -> tuple[SelectOption, ...]:
     return (
+        SelectOption("driver", "Driver default"),
         SelectOption("required", "Required"),
         SelectOption("optional", "Optional"),
         SelectOption("disabled", "Disabled"),
@@ -84,15 +85,13 @@ def _trino_auth_is_kerberos(config: dict[str, str]) -> bool:
             placeholder="HTTP",
             description="Service principal name; blank uses HTTP for Kerberos",
             visible_when=_trino_auth_is_kerberos,
-            advanced=True,
         ),
         SchemaField(
             name="trino_kerberos_hostname_override",
             label="Kerberos Hostname Override",
             placeholder="trino.example.com",
             description="Hostname used to construct the Kerberos service principal",
             visible_when=_trino_auth_is_kerberos,
-            advanced=True,
         ),
         SchemaField(
             name="trino_kerberos_delegate",
@@ -108,7 +107,7 @@ def _trino_auth_is_kerberos(config: dict[str, str]) -> bool:
             label="Mutual Authentication",
             field_type=FieldType.SELECT,
             options=_get_kerberos_mutual_authentication_options(),
-            default="optional",
+            default="driver",
             visible_when=_trino_auth_is_kerberos,
             advanced=True,
         ),

tests/test_password_prompts.py

@@ -133,7 +133,6 @@ def test_mssql_windows_auth_with_empty_password_no_prompt(self) -> None:
         )
         assert not needs_db_password(config)
 
-
     def test_password_command_set_does_not_need_prompt(self) -> None:
         config = ConnectionConfig(
             name="test",
@@ -224,7 +223,6 @@ def test_ssh_password_auth_with_stored_password_does_not_need_prompt(self) -> No
         )
         assert not needs_ssh_password(config)
 
-
     def test_ssh_password_command_set_does_not_need_prompt(self) -> None:
         config = ConnectionConfig(
             name="test",
@@ -481,6 +479,25 @@ def test_explicit_password_skips_command(self, mock_run: MagicMock, mock_getpass
         mock_getpass.assert_not_called()
         assert result.password == "explicit"
 
+    @patch("sqlit.domains.connections.cli.prompts.getpass.getpass")
+    @patch("sqlit.domains.connections.cli.prompts.run_password_command")
+    def test_trino_ticket_auth_skips_stale_password_command(self, mock_run: MagicMock, mock_getpass: MagicMock) -> None:
+        config = ConnectionConfig(
+            name="trino",
+            db_type="trino",
+            server="trino.example.com",
+            username="user",
+            password=None,
+            password_command="echo should-not-run",
+            options={"trino_auth_method": "kerberos"},
+        )
+
+        result = prompt_for_password(config)
+
+        mock_run.assert_not_called()
+        mock_getpass.assert_not_called()
+        assert result.password is None
+
 
 class TestPasswordPromptIntegration:
     """Integration tests for the full password prompt flow."""

tests/unit/test_connection_config_from_dict.py

@@ -205,3 +205,17 @@ def test_to_dict_include_passwords_false_keeps_password_command() -> None:
     d = config.to_dict(include_passwords=False)
     assert d["endpoint"]["password"] is None
     assert d["endpoint"]["password_command"] == "echo pw"
+
+
+def test_transport_endpoint_rewrite_preserves_original_host_without_serializing_it() -> None:
+    config = ConnectionConfig.from_dict({
+        "name": "t",
+        "db_type": "trino",
+        "endpoint": {"kind": "tcp", "host": "trino.example.com", "port": "8443", "username": "u"},
+    })
+
+    tunneled = config.with_endpoint(host="127.0.0.1", port="12345")
+
+    assert tunneled.tcp_endpoint is not None
+    assert tunneled.tcp_endpoint.original_host == "trino.example.com"
+    assert "original_host" not in tunneled.to_dict()["endpoint"]

tests/unit/test_connection_flow.py

@@ -46,3 +46,23 @@ def test_keyring_password_wins_over_command(self, mock_run: MagicMock) -> None:
         mock_run.assert_not_called()
         assert config.tcp_endpoint is not None
         assert config.tcp_endpoint.password == "keyring_pw"
+
+    @patch("sqlit.domains.connections.app.connection_flow.run_password_command")
+    def test_ticket_auth_skips_keyring_and_password_command(self, mock_run: MagicMock) -> None:
+        flow = self._make_flow(keyring_password="stale-keyring-password")
+        config = ConnectionConfig(
+            name="trino",
+            db_type="trino",
+            server="trino.example.com",
+            username="user",
+            password=None,
+            password_command="echo should-not-run",
+            options={"trino_auth_method": "gssapi"},
+        )
+
+        flow.populate_credentials_if_missing(config)
+
+        flow.services.credentials_service.get_password.assert_not_called()
+        mock_run.assert_not_called()
+        assert config.tcp_endpoint is not None
+        assert config.tcp_endpoint.password is None

tests/unit/test_trino_adapter.py

@@ -70,16 +70,45 @@ def test_trino_kerberos_authentication_passes_selected_options():
 
     auth.KerberosAuthentication.assert_called_once_with(
         delegate=True,
-        mutual_authentication=auth.KerberosAuthentication.MUTUAL_OPTIONAL,
         service_name="trino",
         hostname_override="coordinator.example.com",
     )
     assert dbapi.connect.call_args.kwargs["auth"] is auth.KerberosAuthentication.return_value
 
 
-def test_trino_gssapi_rejects_service_name_without_hostname_override():
-    with pytest.raises(ValueError, match="requires a hostname override"):
-        _connect(_config(options={"trino_auth_method": "gssapi", "trino_kerberos_service_name": "trino"}))
+def test_trino_gssapi_defaults_to_http_service_and_endpoint_hostname():
+    dbapi, auth = _connect(_config(options={"trino_auth_method": "gssapi"}))
+
+    auth.GSSAPIAuthentication.assert_called_once_with(
+        delegate=False,
+        service_name="HTTP",
+        hostname_override="trino.example.com",
+    )
+    assert dbapi.connect.call_args.kwargs["auth"] is auth.GSSAPIAuthentication.return_value
+
+
+def test_trino_gssapi_preserves_remote_hostname_across_tunnel_rewrite():
+    config = _config(options={"trino_auth_method": "gssapi"})
+    tunneled_config = config.with_endpoint(host="127.0.0.1", port="12345")
+
+    _, auth = _connect(tunneled_config)
+
+    auth.GSSAPIAuthentication.assert_called_once_with(
+        delegate=False,
+        service_name="HTTP",
+        hostname_override="trino.example.com",
+    )
+    assert "trino_kerberos_hostname_override" not in config.options
+
+
+def test_trino_explicit_mutual_authentication_overrides_driver_default():
+    _, auth = _connect(_config(options={"trino_auth_method": "kerberos", "trino_kerberos_mutual_authentication": "optional"}))
+
+    auth.KerberosAuthentication.assert_called_once_with(
+        delegate=False,
+        mutual_authentication=auth.KerberosAuthentication.MUTUAL_OPTIONAL,
+        service_name="HTTP",
+    )
 
 
 def test_trino_kerberos_missing_extra_opens_package_setup():
@@ -102,7 +131,6 @@ def test_trino_url_uses_kerberos_options_without_passing_them_to_driver():
 
     auth.KerberosAuthentication.assert_called_once_with(
         delegate=False,
-        mutual_authentication=auth.KerberosAuthentication.MUTUAL_OPTIONAL,
         service_name="trino",
         hostname_override="coordinator.example.com",
     )
@@ -127,6 +155,10 @@ def test_trino_schema_exposes_kerberos_authentication_options():
 
     assert [option.value for option in fields["trino_auth_method"].options] == ["none", "basic", "kerberos", "gssapi"]
     assert fields["trino_auth_method"].default == "basic"
-    assert fields["trino_kerberos_mutual_authentication"].default == "optional"
+    assert fields["trino_kerberos_mutual_authentication"].default == "driver"
     assert fields["password"].visible_when is not None
     assert fields["trino_kerberos_service_name"].visible_when is not None
+    assert not fields["trino_kerberos_service_name"].advanced
+    assert not fields["trino_kerberos_hostname_override"].advanced
+    assert fields["trino_kerberos_delegate"].advanced
+    assert fields["trino_kerberos_mutual_authentication"].advanced