fix: use proper OIDC host for China partition

Author: Luis Garza

redshift_connector/plugin/browser_idc_auth_plugin.py

@@ -51,6 +51,7 @@ class OAuthParamNames(Enum):
  CURRENT_INTERACTION_SCHEMA = "https"
  OIDC_SCHEMA = "oidc"
  AMAZON_COM_SCHEMA = "amazonaws.com"
+ AMAZON_CHINA_SCHEMA = "amazonaws.com.cn"
  REDSHIFT_IDC_CONNECT_SCOPE = "redshift:connect"
  AUTH_CODE_GRANT_TYPE = "authorization_code"
  REDIRECT_URI = "http://127.0.0.1"
@@ -389,7 +390,7 @@ def get_authorization_token_url(
  }
 
  encoded_params: str = urlencode(params)
- idc_host = self.OIDC_SCHEMA + "." + str(self.idc_region) + "." + self.AMAZON_COM_SCHEMA
+ idc_host = self._build_oidc_host_url(self.idc_region)
 
  return urlunsplit(
  (
@@ -401,6 +402,31 @@ def get_authorization_token_url(
  )
  )
 
+ def _build_oidc_host_url(self: "BrowserIdcAuthPlugin", idc_region: str) -> str:
+ """
+ Builds the OpenID Connect (OIDC) host URL based on the provided region.
+ 
+ :param idc_region: The AWS region identifier (e.g., "us-west-2", "cn-north-1")
+ :return: The formatted OIDC host URL
+ :raises InterfaceError: if the region is null, empty, or not a valid AWS region
+ """
+ if not idc_region:
+ raise InterfaceError("Region cannot be null or empty")
+ 
+ normalized_region = idc_region.strip().lower()
+ if not normalized_region:
+ raise InterfaceError("Region cannot be empty")
+ 
+ # Validate region format to prevent injection attacks
+ import re
+ # AWS region format: partition-region-az (e.g., us-west-2, cn-north-1, us-gov-east-1)
+ region_pattern = r'^[a-z]{2,3}(-[a-z]+)+-[0-9]+$'
+ if not re.match(region_pattern, normalized_region):
+ raise InterfaceError(f"Invalid AWS region format: {normalized_region}")
+ 
+ domain = self.AMAZON_CHINA_SCHEMA if normalized_region.startswith("cn-") else self.AMAZON_COM_SCHEMA
+ return f"{self.OIDC_SCHEMA}.{normalized_region}.{domain}"
+
  def get_listen_socket(self: "BrowserIdcAuthPlugin", listen_port: int) -> socket.socket:
  """
  Returns a listen socket used for user authentication

test/unit/plugin/test_browser_idc_auth_plugin.py

@@ -14,7 +14,7 @@
 
 def make_valid_browser_idc_provider() -> typing.Tuple[BrowserIdcAuthPlugin, RedshiftProperty]:
  rp: RedshiftProperty = RedshiftProperty()
- rp.idc_region = "some_region"
+ rp.idc_region = "us-west-2"
  rp.issuer_url = "some_url"
  rp.idp_response_timeout = 100
  rp.listen_port = 8000
@@ -25,7 +25,7 @@ def make_valid_browser_idc_provider() -> typing.Tuple[BrowserIdcAuthPlugin, Reds
 
 def valid_browser_without_optional_parameter() -> typing.Tuple[BrowserIdcAuthPlugin, RedshiftProperty]:
  rp: RedshiftProperty = RedshiftProperty()
- rp.idc_region = "some_region"
+ rp.idc_region = "us-west-2"
  rp.issuer_url = "some_url"
  cp: BrowserIdcAuthPlugin = BrowserIdcAuthPlugin()
  cp.add_parameter(rp)
@@ -261,7 +261,7 @@ def test_authorization_token_url():
  mocked_state: str = "mockedState"
  mocked_client_id: str = "mockedClientId"
  mocked_code_challenge: str = "mockedCodeChallenge"
- expected_url = "https://oidc.some_region.amazonaws.com/authorize?response_type=code&client_id=mockedClientId&redirect_uri=None&state=mockedState&scopes=redshift%3Aconnect&code_challenge=mockedCodeChallenge&code_challenge_method=S256"
+ expected_url = "https://oidc.us-west-2.amazonaws.com/authorize?response_type=code&client_id=mockedClientId&redirect_uri=None&state=mockedState&scopes=redshift%3Aconnect&code_challenge=mockedCodeChallenge&code_challenge_method=S256"
 
  url: str = idc_credentials_provider.get_authorization_token_url(
  mocked_state, mocked_client_id, mocked_code_challenge
@@ -292,3 +292,42 @@ def test_open_browser():
 
  listen_socket: socket.socket = idc_credentials_provider.get_listen_socket(mocked_port)
  assert str(listen_socket.getsockname()) == expected_socket
+
+
+@pytest.mark.parametrize("region,expected_host", [
+ ("cn-north-1", "oidc.cn-north-1.amazonaws.com.cn"),
+ ("CN-NORTH-1", "oidc.cn-north-1.amazonaws.com.cn"),
+ (" cn-north-1 ", "oidc.cn-north-1.amazonaws.com.cn"), # Test with spaces
+ ("us-west-2", "oidc.us-west-2.amazonaws.com"),
+ (" US-WEST-2 ", "oidc.us-west-2.amazonaws.com"), # Test with upper case and spaces
+ ("us-gov-west-1", "oidc.us-gov-west-1.amazonaws.com"),
+ ("us-gov-east-1", "oidc.us-gov-east-1.amazonaws.com"),
+ ("us-east-1", "oidc.us-east-1.amazonaws.com"),
+ ("eu-west-1", "oidc.eu-west-1.amazonaws.com"), # Dublin
+ ("ap-southeast-2", "oidc.ap-southeast-2.amazonaws.com"), # Multi-word region
+ ("eu-central-1", "oidc.eu-central-1.amazonaws.com"), # Frankfurt
+])
+def test_build_oidc_host_url_valid_regions(region, expected_host):
+ """Test that valid regions produce correct OIDC host URLs"""
+ idc_credentials_provider, rp = make_valid_browser_idc_provider()
+ 
+ result = idc_credentials_provider._build_oidc_host_url(region)
+ assert result == expected_host
+
+
+@pytest.mark.parametrize("invalid_region", [
+ None,
+ "",
+ " ",
+ "invalid-region",
+ "../../etc/passwd",
+ "us-west-2; rm -rf /",
+ "us-west-2.evil.com",
+ "evil.com#us-west-2"
+])
+def test_build_oidc_host_url_invalid_regions(invalid_region):
+ """Test that invalid regions raise InterfaceError"""
+ idc_credentials_provider, rp = make_valid_browser_idc_provider()
+ 
+ with pytest.raises(InterfaceError):
+ idc_credentials_provider._build_oidc_host_url(invalid_region)