[issue-282] Add Trino Kerberos authentication

README.md

@@ -293,9 +293,24 @@ Most of the time you can just run `sqlit` and connect. If a Python driver is mis
 | Spanner                             | `google-cloud-spanner`       | `pipx inject sqlit-tui google-cloud-spanner`       | `python -m pip install google-cloud-spanner`       |
 | Apache Arrow Flight SQL             | `adbc-driver-flightsql`      | `pipx inject sqlit-tui adbc-driver-flightsql`      | `python -m pip install adbc-driver-flightsql`      |
 | Apache Impala                       | `impyla`                     | `pipx inject sqlit-tui impyla`                     | `python -m pip install impyla`                     |
+| Trino                               | `trino`                      | `pipx inject sqlit-tui trino`                      | `python -m pip install trino`                      |
 | SurrealDB                           | `surrealdb`                  | `pipx inject sqlit-tui surrealdb`                  | `python -m pip install surrealdb`                  |
 | osquery                             | `osquery`                    | `pipx inject sqlit-tui osquery`                    | `python -m pip install osquery`                    |
 
+### Trino Kerberos Authentication
+
+To connect to a Trino server with an existing Kerberos ticket, install the matching Trino authentication extra before launching sqlit:
+
+```bash
+# pipx installation
+pipx inject sqlit-tui 'trino[kerberos]'
+
+# pip or virtual environment installation
+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.
+
 ### SSH Tunnel Support
 
 SSH tunnel functionality requires additional dependencies. Install with the `ssh` extra:

pyproject.toml

@@ -73,6 +73,8 @@ db2 = ["ibm_db>=3.2.0"]
 hana = ["hdbcli>=2.20.0"]
 teradata = ["teradatasql>=20.0.0"]
 trino = ["trino>=0.329.0"]
+trino-kerberos = ["trino[kerberos]>=0.329.0"]
+trino-gssapi = ["trino[gssapi]>=0.329.0"]
 presto = ["presto-python-client>=0.8.4"]
 bigquery = ["google-cloud-bigquery"]
 spanner = ["google-cloud-spanner>=3.0.0"]

sqlit/domains/connections/app/url_parser.py

@@ -132,9 +132,18 @@ def parse_connection_url(
     if schema.is_file_based:
         strategy = FILE_BASED_STRATEGY
 
-    return normalize_connection_config(
-        strategy.parse(parsed, db_type, config_name, url, extra_options)
-    )
+    config = strategy.parse(parsed, db_type, config_name, url, extra_options)
+    if db_type == "trino":
+        for name in (
+            "trino_auth_method",
+            "trino_kerberos_delegate",
+            "trino_kerberos_hostname_override",
+            "trino_kerberos_mutual_authentication",
+            "trino_kerberos_service_name",
+        ):
+            if name in config.extra_options:
+                config.set_option(name, config.extra_options.pop(name))
+    return normalize_connection_config(config)
 
 
 def _parse_file_based_url(

sqlit/domains/connections/cli/prompts.py

@@ -32,6 +32,10 @@ def _needs_db_prompt(config: ConnectionConfig) -> bool:
     auth_type = config.get_option("auth_type")
     if auth_type in ("ad_default", "ad_integrated", "windows"):
         return False
+    if config.db_type == "trino":
+        auth_method = str(config.options.get("trino_auth_method", config.extra_options.get("trino_auth_method", "basic"))).lower()
+        if auth_method in {"none", "kerberos", "gssapi"}:
+            return False
     endpoint = config.tcp_endpoint
     return bool(endpoint and endpoint.password is None)
 

sqlit/domains/connections/domain/passwords.py

@@ -18,6 +18,11 @@ def needs_db_password(config: ConnectionConfig) -> bool:
     if auth_type in ("ad_default", "ad_integrated", "windows"):
         return False
 
+    if config.db_type == "trino":
+        auth_method = str(config.options.get("trino_auth_method", config.extra_options.get("trino_auth_method", "basic"))).lower()
+        if auth_method in {"none", "kerberos", "gssapi"}:
+            return False
+
     endpoint = config.tcp_endpoint
     if not endpoint or endpoint.password is not None:
         return False

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

@@ -21,6 +21,14 @@
 class TrinoAdapter(CursorBasedAdapter):
     """Adapter for Trino (PrestoSQL) using trino client."""
 
+    _AUTH_OPTION_NAMES = {
+        "trino_auth_method",
+        "trino_kerberos_delegate",
+        "trino_kerberos_hostname_override",
+        "trino_kerberos_mutual_authentication",
+        "trino_kerberos_service_name",
+    }
+
     @property
     def name(self) -> str:
         return "Trino"
@@ -99,15 +107,95 @@ def connect(self, config: ConnectionConfig) -> Any:
         if schema:
             connect_args["schema"] = schema
 
-        if endpoint.password:
+        auth = self._build_authentication(config, endpoint.username, endpoint.password)
+        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:
+        default_method = "basic" if password else "none"
+        auth_method = str(self._get_authentication_option(config, "trino_auth_method", default_method)).lower()
+
+        if auth_method == "none":
+            return None
+
+        if auth_method == "basic":
+            if not password:
+                return None
             try:
                 from trino.auth import BasicAuthentication
             except Exception as exc:
                 raise ValueError("Trino password authentication requires trino.auth.BasicAuthentication") from exc
-            connect_args["auth"] = BasicAuthentication(endpoint.username, endpoint.password)
-
-        connect_args.update(config.extra_options)
-        return trino_dbapi.connect(**connect_args)
+            return BasicAuthentication(username, password)
+
+        if auth_method not in {"kerberos", "gssapi"}:
+            raise ValueError(f"Unsupported Trino authentication method: {auth_method}")
+
+        package_extra = auth_method
+        try:
+            if auth_method == "kerberos":
+                from requests_kerberos import (  # type: ignore[import-untyped]
+                    DISABLED,
+                    OPTIONAL,
+                    REQUIRED,
+                )
+                from trino.auth import KerberosAuthentication
+
+                auth_class = KerberosAuthentication
+            else:
+                from requests_gssapi import (  # type: ignore[import-untyped]
+                    DISABLED,
+                    OPTIONAL,
+                    REQUIRED,
+                )
+                from trino.auth import GSSAPIAuthentication
+
+                auth_class = GSSAPIAuthentication
+        except ImportError as exc:
+            from sqlit.domains.connections.providers.exceptions import MissingDriverError
+
+            raise MissingDriverError(
+                f"Trino {auth_method.upper()} authentication",
+                f"trino-{package_extra}",
+                f"trino[{package_extra}]",
+                module_name=f"requests_{package_extra}",
+                import_error=str(exc),
+            ) from exc
+        except Exception as exc:
+            raise ValueError(
+                f"Trino {auth_method.upper()} authentication requires `trino[{package_extra}]`. "
+                f"Install it with `pipx inject sqlit-tui 'trino[{package_extra}]'` or "
+                f"`python -m pip install 'trino[{package_extra}]'`."
+            ) from exc
+
+        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_values = {
+            "required": REQUIRED,
+            "optional": OPTIONAL,
+            "disabled": DISABLED,
+        }
+        if 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]
+        service_name = self._get_authentication_option(config, "trino_kerberos_service_name")
+        if auth_method == "kerberos" and 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 service_name:
+            auth_args["service_name"] = str(service_name)
+        if hostname_override:
+            auth_args["hostname_override"] = str(hostname_override)
+        return auth_class(**auth_args)
+
+    def _get_authentication_option(self, config: ConnectionConfig, name: str, default: Any = None) -> Any:
+        return config.options.get(name, config.extra_options.get(name, default))
 
     def get_databases(self, conn: Any) -> list[str]:
         cursor = conn.cursor()

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

@@ -6,7 +6,6 @@
     FieldType,
     SchemaField,
     SelectOption,
-    _password_field,
     _port_field,
     _server_field,
     _username_field,
@@ -20,6 +19,31 @@ def _get_http_scheme_options() -> tuple[SelectOption, ...]:
     )
 
 
+def _get_authentication_options() -> tuple[SelectOption, ...]:
+    return (
+        SelectOption("none", "None"),
+        SelectOption("basic", "Basic"),
+        SelectOption("kerberos", "Kerberos"),
+        SelectOption("gssapi", "GSSAPI"),
+    )
+
+
+def _get_kerberos_mutual_authentication_options() -> tuple[SelectOption, ...]:
+    return (
+        SelectOption("required", "Required"),
+        SelectOption("optional", "Optional"),
+        SelectOption("disabled", "Disabled"),
+    )
+
+
+def _trino_auth_is_basic(config: dict[str, str]) -> bool:
+    return config.get("trino_auth_method", "basic") == "basic"
+
+
+def _trino_auth_is_kerberos(config: dict[str, str]) -> bool:
+    return config.get("trino_auth_method", "basic") in {"kerberos", "gssapi"}
+
+
 SCHEMA = ConnectionSchema(
     db_type="trino",
     display_name="Trino",
@@ -39,7 +63,55 @@ def _get_http_scheme_options() -> tuple[SelectOption, ...]:
             required=False,
         ),
         _username_field(),
-        _password_field(),
+        SchemaField(
+            name="trino_auth_method",
+            label="Authentication",
+            field_type=FieldType.SELECT,
+            options=_get_authentication_options(),
+            default="basic",
+        ),
+        SchemaField(
+            name="password",
+            label="Password",
+            field_type=FieldType.PASSWORD,
+            placeholder="(empty = ask every connect)",
+            group="credentials",
+            visible_when=_trino_auth_is_basic,
+        ),
+        SchemaField(
+            name="trino_kerberos_service_name",
+            label="Kerberos Service Name",
+            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",
+            label="Delegate Kerberos Credentials",
+            field_type=FieldType.SELECT,
+            options=(SelectOption("false", "No"), SelectOption("true", "Yes")),
+            default="false",
+            visible_when=_trino_auth_is_kerberos,
+            advanced=True,
+        ),
+        SchemaField(
+            name="trino_kerberos_mutual_authentication",
+            label="Mutual Authentication",
+            field_type=FieldType.SELECT,
+            options=_get_kerberos_mutual_authentication_options(),
+            default="optional",
+            visible_when=_trino_auth_is_kerberos,
+            advanced=True,
+        ),
         SchemaField(
             name="http_scheme",
             label="HTTP Scheme",

sqlit/domains/connections/ui/screens/connection.py

@@ -27,6 +27,7 @@
     DatabaseType,
     get_database_type_labels,
 )
+from sqlit.domains.connections.domain.passwords import needs_db_password
 from sqlit.domains.connections.providers.catalog import get_provider_schema
 from sqlit.domains.connections.providers.driver import ensure_provider_driver_available
 from sqlit.domains.connections.providers.exceptions import MissingDriverError
@@ -157,8 +158,8 @@ def _get_field_container(self, field_name: str) -> Container | None:
 
     def _on_browse_file(self, field_name: str) -> None:
         """Open file picker for a file field."""
-        from sqlit.shared.ui.screens.file_picker import FilePickerMode, FilePickerScreen
         from sqlit.domains.connections.ui.fields import FieldType
+        from sqlit.shared.ui.screens.file_picker import FilePickerMode, FilePickerScreen
 
         # Get current value from the field
         current_value = ""
@@ -711,8 +712,7 @@ def on_ssh_password(password: str | None) -> None:
             )
             return
 
-        endpoint = config.tcp_endpoint
-        if not is_file_based(config.db_type) and endpoint and endpoint.password is None:
+        if needs_db_password(config):
 
             def on_db_password(password: str | None) -> None:
                 if password is None: