Repository: dbt-labs/dbt-redshift (python)

Issue:
[ADAP-512] [Regression] Can't connect to Redshift with dbt-redshift 1.5 due to SSLCertVerificationError
### Is this a regression in a recent version of dbt-redshift?

- [X] I believe this is a regression in dbt-redshift functionality
- [X] I have searched the existing issues, and I could not find an existing issue for this regression

### Current Behavior

After upgrading to dbt 1.5.1, we are unable to connect to Redshift due to an SSLCertVerificationError. Our Redshift connections are made through StrongDM, which may be affecting the certification? I've been able to connect with `redshift_connector` with `ssl=False`, so a quick fix would be to allow passing the `ssl` connection parameter in via the profile.yml.

### Expected/Previous Behavior

Connection to Redshift works as it did before.

### Steps To Reproduce

1. Use StrongDM to connect to Redshift on your local machine.
2. Set up profiles.yml to point to the relevant port on localhost.
3. Try to run any dbt command that requires a database connection

### Relevant log output

```shell
❯ dbt debug
19:54:08  Running with dbt=1.5.0
19:54:08  dbt version: 1.5.0
19:54:08  python version: 3.9.16
19:54:08  python path: /Users/shirley.zhang/.conda/envs/dbt-1.5/bin/python
19:54:08  os info: macOS-10.16-x86_64-i386-64bit
19:54:08  Using profiles.yml file at /Users/shirley.zhang/.dbt/profiles.yml
19:54:08  Using dbt_project.yml file at /Users/shirley.zhang/repos/dbt/ds-dbt-upgrade/dbt_project.yml
19:54:08  Configuration:
19:54:09    profiles.yml file [OK found and valid]
19:54:09    dbt_project.yml file [OK found and valid]
19:54:09  Required dependencies:
19:54:09   - git [OK found]

19:54:09  Connection:
19:54:09    host: localhost
19:54:09    port: 5539
19:54:09    user: 
19:54:09    database: XXXXX
19:54:09    schema: user__shirley
19:54:09    method: database
19:54:09    cluster_id: None
19:54:09    iam_profile: None
19:54:09    sslmode: None
19:54:09    region: us-east-1
19:54:09    Connection test: [ERROR]

19:54:09  1 check failed:
19:54:09  dbt was unable to connect to the specified database.
The database returned the following error:

  >Database Error
  ('communication error', SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1129)'))

Check your database credentials and try again. For more information, visit:
https://docs.getdbt.com/docs/configure-your-profile
```


### Environment

```markdown
- OS: MacOS 10.16
- Python: 3.9.16
- dbt-core (working version): 1.5.0
- dbt-redshift (working version): 1.5.1
- dbt-core (regression version): 1.4.6
- dbt-redshift (regression version): 1.4.0
```


### Additional Context

Using `redshift_connector` directly in python, full traceback for the connection error
```
>>> c = redshift_connector.connect(user='', password='', host='localhost', database='XXXX', port=5539)
Traceback (most recent call last):
  File "/Users/shirley.zhang/.conda/envs/dbt-1.5/lib/python3.9/site-packages/redshift_connector/core.py", line 619, in __init__
    self._usock = ssl_context.wrap_socket(self._usock)
  File "/Users/shirley.zhang/.conda/envs/dbt-1.5/lib/python3.9/ssl.py", line 501, in wrap_socket
    return self.sslsocket_class._create(
  File "/Users/shirley.zhang/.conda/envs/dbt-1.5/lib/python3.9/ssl.py", line 1041, in _create
    self.do_handshake()
  File "/Users/shirley.zhang/.conda/envs/dbt-1.5/lib/python3.9/ssl.py", line 1310, in do_handshake
    self._sslobj.do_handshake()
ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1129)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/shirley.zhang/.conda/envs/dbt-1.5/lib/python3.9/site-packages/redshift_connector/__init__.py", line 340, in connect
    return Connection(
  File "/Users/shirley.zhang/.conda/envs/dbt-1.5/lib/python3.9/site-packages/redshift_connector/core.py", line 632, in __init__
    raise InterfaceError("communication error", e)
redshift_connector.error.InterfaceError: ('communication error', SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1129)'))
```

Using `ssl=False` works:
```
>>> c = redshift_connector.connect(user='', password='', host='localhost', database='XXXX', port=5539, ssl=False)
>>> c.cursor().execute("select count(*) from pg_user;").fetchall()
([62],)
```

Pull request description:
ADAP-512: sslmode translation
resolves #429

### Description

Translate from the user provided `sslmode` to the parameters required by the redshift connector in order to preserve the `dbt-redshift` config interface after the shift from `psycopg2` to `redshift-connector`.

Please see this PR for a significantly more detailed writeup: https://github.com/dbt-labs/dbt-redshift/pull/439. Thanks @jiezhen-chen for this work and the detail in the write-up. Most of the changes with my name attached revolve around reworking @jiezhen-chen's solution to align with pre-existing patterns in `dbt-core`.

### Checklist

- [x] I have read [the contributing guide](https://github.com/dbt-labs/dbt-redshift/blob/main/CONTRIBUTING.md) and understand what's expected of me
- [x] I have signed the [CLA](https://docs.getdbt.com/docs/contributor-license-agreements)
- [x] I have run this code in development and it appears to resolve the stated issue
- [x] This PR includes tests, or tests are not required/relevant for this PR
- [x] I have [opened an issue to add/update docs](https://github.com/dbt-labs/docs.getdbt.com/issues/new/choose), or docs changes are not required/relevant for this PR
- [x] I have run `changie new` to [create a changelog entry](https://github.com/dbt-labs/dbt-redshift/blob/main/CONTRIBUTING.md#Adding-CHANGELOG-Entry)


Patch:
diff --git a/.changes/unreleased/Breaking Changes-20230522-111945.yaml b/.changes/unreleased/Breaking Changes-20230522-111945.yaml
new file mode 100644
index 00000000..63a03b3b
--- /dev/null
+++ b/.changes/unreleased/Breaking Changes-20230522-111945.yaml
@@ -0,0 +1,7 @@
+kind: Breaking Changes
+body: sslmode behavior has changed. To connect without ssl, set sslmode = disable.
+  To connect using ssl, set sslmode to verify-ca, or verify-full.
+time: 2023-05-22T11:19:45.927903-07:00
+custom:
+  Author: jiezhen-chen
+  Issue: "429"
diff --git a/.changes/unreleased/Fixes-20230512-082027.yaml b/.changes/unreleased/Fixes-20230512-082027.yaml
new file mode 100644
index 00000000..7116257d
--- /dev/null
+++ b/.changes/unreleased/Fixes-20230512-082027.yaml
@@ -0,0 +1,6 @@
+kind: Fixes
+body: translate psycopg2 sslmode to ssl and sslmode in redshift_connector
+time: 2023-05-12T08:20:27.486301-07:00
+custom:
+  Author: jiezhen-chen
+  Issue: "429"
diff --git a/dbt/adapters/redshift/connections.py b/dbt/adapters/redshift/connections.py
index 1a1e7353..509fd491 100644
--- a/dbt/adapters/redshift/connections.py
+++ b/dbt/adapters/redshift/connections.py
@@ -13,12 +13,25 @@ from redshift_connector.utils.oids import get_datatype_name
 
 from dbt.adapters.sql import SQLConnectionManager
 from dbt.contracts.connection import AdapterResponse, Connection, Credentials
+from dbt.contracts.util import Replaceable
+from dbt.dataclass_schema import FieldEncoder, dbtClassMixin, StrEnum, ValidationError
 from dbt.events import AdapterLogger
-import dbt.exceptions
+from dbt.exceptions import DbtRuntimeError, CompilationError
 import dbt.flags
-from dbt.dataclass_schema import FieldEncoder, dbtClassMixin, StrEnum
 from dbt.helper_types import Port
 
+
+class SSLConfigError(CompilationError):
+    def __init__(self, exc: ValidationError):
+        self.exc = exc
+        super().__init__(msg=self.get_message())
+
+    def get_message(self) -> str:
+        validator_msg = self.validator_error_message(self.exc)
+        msg = f"Could not parse SSL config: {validator_msg}"
+        return msg
+
+
 logger = AdapterLogger("Redshift")
 
 drop_lock: Lock = dbt.flags.MP_CONTEXT.Lock()  # type: ignore
@@ -57,6 +70,66 @@ class RedshiftConnectionMethod(StrEnum):
     IAM = "iam"
 
 
+class UserSSLMode(StrEnum):
+    disable = "disable"
+    allow = "allow"
+    prefer = "prefer"
+    require = "require"
+    verify_ca = "verify-ca"
+    verify_full = "verify-full"
+
+    @classmethod
+    def default(cls) -> "UserSSLMode":
+        # default for `psycopg2`, which aligns with dbt-redshift 1.4 and provides backwards compatibility
+        return cls.prefer
+
+
+class RedshiftSSLMode(StrEnum):
+    verify_ca = "verify-ca"
+    verify_full = "verify-full"
+
+
+SSL_MODE_TRANSLATION = {
+    UserSSLMode.disable: None,
+    UserSSLMode.allow: RedshiftSSLMode.verify_ca,
+    UserSSLMode.prefer: RedshiftSSLMode.verify_ca,
+    UserSSLMode.require: RedshiftSSLMode.verify_ca,
+    UserSSLMode.verify_ca: RedshiftSSLMode.verify_ca,
+    UserSSLMode.verify_full: RedshiftSSLMode.verify_full,
+}
+
+
+@dataclass
+class RedshiftSSLConfig(dbtClassMixin, Replaceable):  # type: ignore
+    ssl: bool = True
+    sslmode: Optional[RedshiftSSLMode] = SSL_MODE_TRANSLATION[UserSSLMode.default()]
+
+    @classmethod
+    def parse(cls, user_sslmode: UserSSLMode) -> "RedshiftSSLConfig":
+        try:
+            raw_redshift_ssl = {
+                "ssl": user_sslmode != UserSSLMode.disable,
+                "sslmode": SSL_MODE_TRANSLATION[user_sslmode],
+            }
+            cls.validate(raw_redshift_ssl)
+        except ValidationError as exc:
+            raise SSLConfigError(exc)
+
+        redshift_ssl = cls.from_dict(raw_redshift_ssl)
+
+        if redshift_ssl.ssl:
+            message = (
+                f"Establishing connection using ssl with `sslmode` set to '{user_sslmode}'."
+                f"To connect without ssl, set `sslmode` to 'disable'."
+            )
+        else:
+            message = "Establishing connection without ssl."
+
+        logger.debug(message)
+
+        return redshift_ssl
+
+
 @dataclass
 class RedshiftCredentials(Credentials):
     host: str
@@ -74,7 +147,7 @@ class RedshiftCredentials(Credentials):
     ra3_node: Optional[bool] = False
     connect_timeout: Optional[int] = None
     role: Optional[str] = None
-    sslmode: Optional[str] = None
+    sslmode: Optional[UserSSLMode] = field(default_factory=UserSSLMode.default)
     retries: int = 1
     region: Optional[str] = None  # if not provided, will be determined from host
     autocommit: Optional[bool] = False
@@ -146,8 +219,8 @@ class RedshiftConnectMethodFactory:
                 "Invalid region provided: {}".format(kwargs["region"])
             )
 
-        if self.credentials.sslmode:
-            kwargs["sslmode"] = self.credentials.sslmode
+        redshift_ssl_config = RedshiftSSLConfig.parse(self.credentials.sslmode)
+        kwargs.update(redshift_ssl_config.to_dict())
 
         # Support missing 'method' for backwards compatibility
         if method == RedshiftConnectionMethod.DATABASE or method is None:
@@ -342,7 +415,7 @@ class RedshiftConnectionManager(SQLConnectionManager):
         if cursor is None:
             conn = self.get_thread_connection()
             conn_name = conn.name if conn and conn.name else "<None>"
-            raise dbt.exceptions.DbtRuntimeError(f"Tried to run invalid SQL: {sql} on {conn_name}")
+            raise DbtRuntimeError(f"Tried to run invalid SQL: {sql} on {conn_name}")
 
         return connection, cursor
 


Test patch:
diff --git a/tests/unit/test_redshift_adapter.py b/tests/unit/test_redshift_adapter.py
index 9a43c1ce..5e4e00b9 100644
--- a/tests/unit/test_redshift_adapter.py
+++ b/tests/unit/test_redshift_adapter.py
@@ -12,7 +12,7 @@ from dbt.adapters.redshift import (
 )
 from dbt.clients import agate_helper
 from dbt.exceptions import FailedToConnectError
-from dbt.adapters.redshift.connections import RedshiftConnectMethodFactory
+from dbt.adapters.redshift.connections import RedshiftConnectMethodFactory, RedshiftSSLConfig
 from .utils import (
     config_from_parts_or_dicts,
     mock_connection,
@@ -21,6 +21,9 @@ from .utils import (
 )
 
 
+DEFAULT_SSL_CONFIG = RedshiftSSLConfig().to_dict()
+
+
 class TestRedshiftAdapter(unittest.TestCase):
     def setUp(self):
         profile_cfg = {
@@ -74,6 +77,7 @@ class TestRedshiftAdapter(unittest.TestCase):
             db_groups=[],
             timeout=None,
             region="us-east-1",
+            **DEFAULT_SSL_CONFIG,
         )
 
     @mock.patch("redshift_connector.connect", Mock())
@@ -92,6 +96,7 @@ class TestRedshiftAdapter(unittest.TestCase):
             db_groups=[],
             region="us-east-1",
             timeout=None,
+            **DEFAULT_SSL_CONFIG,
         )
 
     @mock.patch("redshift_connector.connect", Mock())
@@ -112,11 +117,12 @@ class TestRedshiftAdapter(unittest.TestCase):
             user="",
             cluster_identifier="my_redshift",
             region="us-east-1",
+            timeout=None,
             auto_create=False,
             db_groups=[],
             profile=None,
-            timeout=None,
             port=5439,
+            **DEFAULT_SSL_CONFIG,
         )
 
     @mock.patch("redshift_connector.connect", Mock())
@@ -134,6 +140,7 @@ class TestRedshiftAdapter(unittest.TestCase):
             db_groups=[],
             region="us-east-1",
             timeout=30,
+            **DEFAULT_SSL_CONFIG,
         )
 
     @mock.patch("redshift_connector.connect", Mock())
@@ -162,6 +169,7 @@ class TestRedshiftAdapter(unittest.TestCase):
             profile="test",
             timeout=None,
             port=5439,
+            **DEFAULT_SSL_CONFIG,
         )
 
     @mock.patch("redshift_connector.connect", Mock())
@@ -188,6 +196,7 @@ class TestRedshiftAdapter(unittest.TestCase):
             profile="test",
             timeout=None,
             port=5439,
+            **DEFAULT_SSL_CONFIG,
         )
 
     @mock.patch("redshift_connector.connect", Mock())
@@ -216,6 +225,7 @@ class TestRedshiftAdapter(unittest.TestCase):
             profile="test",
             timeout=None,
             port=5439,
+            **DEFAULT_SSL_CONFIG,
         )
 
     @mock.patch("redshift_connector.connect", Mock())
@@ -245,6 +255,7 @@ class TestRedshiftAdapter(unittest.TestCase):
                 profile="test",
                 timeout=None,
                 port=5439,
+                **DEFAULT_SSL_CONFIG,
             )
 
     @mock.patch("redshift_connector.connect", Mock())
@@ -274,8 +285,104 @@ class TestRedshiftAdapter(unittest.TestCase):
                 profile="test",
                 timeout=None,
                 port=5439,
+                **DEFAULT_SSL_CONFIG,
             )
 
+    @mock.patch("redshift_connector.connect", Mock())
+    def test_sslmode_disable(self):
+        self.config.credentials.sslmode = "disable"
+        connection = self.adapter.acquire_connection("dummy")
+        connection.handle
+        redshift_connector.connect.assert_called_once_with(
+            host="thishostshouldnotexist.test.us-east-1",
+            database="redshift",
+            user="root",
+            password="password",
+            port=5439,
+            auto_create=False,
+            db_groups=[],
+            region="us-east-1",
+            timeout=None,
+            ssl=False,
+            sslmode=None,
+        )
+
+    @mock.patch("redshift_connector.connect", Mock())
+    def test_sslmode_allow(self):
+        self.config.credentials.sslmode = "allow"
+        connection = self.adapter.acquire_connection("dummy")
+        connection.handle
+        redshift_connector.connect.assert_called_once_with(
+            host="thishostshouldnotexist.test.us-east-1",
+            database="redshift",
+            user="root",
+            password="password",
+            port=5439,
+            auto_create=False,
+            db_groups=[],
+            region="us-east-1",
+            timeout=None,
+            ssl=True,
+            sslmode="verify-ca",
+        )
+
+    @mock.patch("redshift_connector.connect", Mock())
+    def test_sslmode_verify_full(self):
+        self.config.credentials.sslmode = "verify-full"
+        connection = self.adapter.acquire_connection("dummy")
+        connection.handle
+        redshift_connector.connect.assert_called_once_with(
+            host="thishostshouldnotexist.test.us-east-1",
+            database="redshift",
+            user="root",
+            password="password",
+            port=5439,
+            auto_create=False,
+            db_groups=[],
+            region="us-east-1",
+            timeout=None,
+            ssl=True,
+            sslmode="verify-full",
+        )
+
+    @mock.patch("redshift_connector.connect", Mock())
+    def test_sslmode_verify_ca(self):
+        self.config.credentials.sslmode = "verify-ca"
+        connection = self.adapter.acquire_connection("dummy")
+        connection.handle
+        redshift_connector.connect.assert_called_once_with(
+            host="thishostshouldnotexist.test.us-east-1",
+            database="redshift",
+            user="root",
+            password="password",
+            port=5439,
+            auto_create=False,
+            db_groups=[],
+            region="us-east-1",
+            timeout=None,
+            ssl=True,
+            sslmode="verify-ca",
+        )
+
+    @mock.patch("redshift_connector.connect", Mock())
+    def test_sslmode_prefer(self):
+        self.config.credentials.sslmode = "prefer"
+        connection = self.adapter.acquire_connection("dummy")
+        connection.handle
+        redshift_connector.connect.assert_called_once_with(
+            host="thishostshouldnotexist.test.us-east-1",
+            database="redshift",
+            user="root",
+            password="password",
+            port=5439,
+            auto_create=False,
+            db_groups=[],
+            region="us-east-1",
+            timeout=None,
+            ssl=True,
+            sslmode="verify-ca",
+        )
+
     @mock.patch("redshift_connector.connect", Mock())
     @mock.patch("boto3.Session", Mock())
     def test_serverless_iam_failure(self):
@@ -301,6 +408,7 @@ class TestRedshiftAdapter(unittest.TestCase):
                 profile="test",
                 port=5439,
                 timeout=None,
+                **DEFAULT_SSL_CONFIG,
             )
         self.assertTrue("'host' must be provided" in context.exception.msg)