Add allow_insecure_auth connection arg

README.md

@@ -365,6 +365,41 @@ It follows the interface for `KerberosAuthentication`, but is using
     )
     ```
 
+
+### Allow Insecure Authentication
+
+Authentication attempts are automatically enforced to use HTTPS. Although not recommended, you can enable authentication over HTTP by adding `allow_insecure_auth` connection argument.
+
+- DBAPI
+
+    ```python
+    from trino.dbapi import connect
+    ...
+
+    conn = connect(
+        auth=...,
+        allow_insecure_auth=True,
+        ...
+    )
+    ```
+
+- SQLAlchemy
+
+    ```python
+    from sqlalchemy import create_engine
+
+    engine = create_engine("trino://<username>:<password>@<host>:<port>/<catalog>?allow_insecure_auth=true")
+
+    # or as connect_args
+    engine = create_engine(
+        "trino://<username>@<host>:<port>/<catalog>",
+        connect_args={
+            "auth"=...,
+            "allow_insecure_auth": True,
+        }
+    )
+    ```
+
 ## User impersonation
 
 In the case where user who submits the query is not the same as user who authenticates to Trino server (e.g in Superset),

tests/unit/sqlalchemy/test_dialect.py

@@ -324,3 +324,16 @@ def test_trino_connection_oauth2_auth():
 
     assert cparams['http_scheme'] == "https"
     assert isinstance(cparams['auth'], OAuth2Authentication)
+
+
+def test_trino_connection_allow_insecure_auth():
+    dialect = TrinoDialect()
+    username = 'trino-user'
+    password = 'trino-bunny'
+    url = make_url(f'trino://{username}:{password}@host?allow_insecure_auth=true')
+    _, cparams = dialect.create_connect_args(url)
+
+    assert 'http_scheme' not in cparams
+    assert isinstance(cparams['auth'], BasicAuthentication)
+    assert cparams['auth']._username == username
+    assert cparams['auth']._password == password

tests/unit/test_client.py

@@ -339,6 +339,29 @@ def test_http_scheme_with_port(mock_get_and_post):
     assert parsed_url.port == constants.DEFAULT_TLS_PORT
 
 
+def test_allow_insecure_auth(mock_get_and_post):
+    _, post = mock_get_and_post
+
+    req = TrinoRequest(
+        host="coordinator",
+        port=8080,
+        client_session=ClientSession(
+            user="test",
+        ),
+        http_scheme=constants.HTTP,
+
+        auth=KerberosAuthentication(),
+        allow_insecure_auth=True,
+    )
+
+    req.post("SELECT 1")
+    post_args, _ = post.call_args
+    parsed_url = urlparse(post_args[0])
+
+    assert parsed_url.scheme == constants.HTTP
+    assert parsed_url.port == 8080
+
+
 def test_request_timeout():
     timeout = 0.1
     http_scheme = "http"

trino/client.py

@@ -415,6 +415,7 @@ class TrinoRequest:
     :param http_scheme: "http" or "https"
     :param auth: class that manages user authentication. ``None`` means no
                  authentication.
+    :param allow_insecure_auth: allow insecure authentication over HTTP
     :max_attempts: maximum number of attempts when sending HTTP requests. An
                    attempt is an HTTP request. 5 attempts means 4 retries.
     :request_timeout: How long (in seconds) to wait for the server to send
@@ -462,6 +463,7 @@ def __init__(
         http_session: Optional[Session] = None,
         http_scheme: Optional[str] = None,
         auth: Optional[Authentication] = constants.DEFAULT_AUTH,
+        allow_insecure_auth: bool = False,
         max_attempts: int = MAX_ATTEMPTS,
         request_timeout: Union[float, Tuple[float, float]] = constants.DEFAULT_REQUEST_TIMEOUT,
         handle_retry=_RetryWithExponentialBackoff(),
@@ -488,8 +490,9 @@ def __init__(
         self._http_session.headers.update(self.http_headers)
         self._exceptions = self.HTTP_EXCEPTIONS
         self._auth = auth
+        self._allow_insecure_auth = allow_insecure_auth
         if self._auth:
-            if self._http_scheme == constants.HTTP:
+            if self._http_scheme == constants.HTTP and not self._allow_insecure_auth:
                 raise ValueError("cannot use authentication with HTTP")
             self._auth.set_http_session(self._http_session)
             self._exceptions += self._auth.get_exceptions()

trino/dbapi.py

@@ -152,6 +152,7 @@ def __init__(
         http_headers=None,
         http_scheme=constants.HTTP,
         auth=constants.DEFAULT_AUTH,
+        allow_insecure_auth=False,
         extra_credential=None,
         max_attempts=constants.DEFAULT_MAX_ATTEMPTS,
         request_timeout=constants.DEFAULT_REQUEST_TIMEOUT,
@@ -205,6 +206,7 @@ def __init__(
         self.http_headers = http_headers
         self.http_scheme = http_scheme if not parsed_host.scheme else parsed_host.scheme
         self.auth = auth
+        self.allow_insecure_auth = allow_insecure_auth
         self.extra_credential = extra_credential
         self.max_attempts = max_attempts
         self.request_timeout = request_timeout
@@ -264,6 +266,7 @@ def _create_request(self):
             self._http_session,
             self.http_scheme,
             self.auth,
+            self.allow_insecure_auth,
             self.max_attempts,
             self.request_timeout,
         )

trino/sqlalchemy/dialect.py

@@ -130,22 +130,30 @@ def create_connect_args(self, url: URL) -> Tuple[Sequence[Any], Mapping[str, Any
         if url.username:
             kwargs["user"] = unquote_plus(url.username)
 
+        allow_insecure_auth = "allow_insecure_auth" in url.query
+        if allow_insecure_auth:
+            kwargs["allow_insecure_auth"] = True
+
         if url.password:
             if not url.username:
                 raise ValueError("Username is required when specify password in connection URL")
-            kwargs["http_scheme"] = "https"
+            if not allow_insecure_auth:
+                kwargs["http_scheme"] = "https"
             kwargs["auth"] = BasicAuthentication(unquote_plus(url.username), unquote_plus(url.password))
 
         if "access_token" in url.query:
-            kwargs["http_scheme"] = "https"
+            if not allow_insecure_auth:
+                kwargs["http_scheme"] = "https"
             kwargs["auth"] = JWTAuthentication(unquote_plus(url.query["access_token"]))
 
         if "cert" in url.query and "key" in url.query:
-            kwargs["http_scheme"] = "https"
+            if not allow_insecure_auth:
+                kwargs["http_scheme"] = "https"
             kwargs["auth"] = CertificateAuthentication(unquote_plus(url.query['cert']), unquote_plus(url.query['key']))
 
         if "externalAuthentication" in url.query:
-            kwargs["http_scheme"] = "https"
+            if not allow_insecure_auth:
+                kwargs["http_scheme"] = "https"
             kwargs["auth"] = OAuth2Authentication()
 
         if "source" in url.query: