feat(security_integration): complete custom OAuth field coverage

snowcap/data_provider.py

@@ -3122,17 +3122,35 @@ def fetch_security_integration(session: SnowflakeConnection, fqn: FQN):
                 "enabled": data["enabled"] == "true",
                 "oauth_client_type": properties.get("oauth_client_type"),
                 "oauth_redirect_uri": properties.get("oauth_redirect_uri"),
+                "oauth_allow_non_tls_redirect_uri": properties.get("oauth_allow_non_tls_redirect_uri"),
                 "oauth_issue_refresh_tokens": properties.get("oauth_issue_refresh_tokens"),
                 "oauth_refresh_token_validity": oauth_refresh_token_validity,
+                "oauth_single_use_refresh_tokens_required": properties.get("oauth_single_use_refresh_tokens_required"),
                 "oauth_use_secondary_roles": properties.get("oauth_use_secondary_roles"),
+                "oauth_any_role_mode": properties.get("oauth_any_role_mode"),
                 "oauth_enforce_pkce": properties.get("oauth_enforce_pkce"),
+                # oauth_enable_role_selection is CREATE-only (DESC never returns it); it is
+                # marked unfetchable on the spec, so it is intentionally not read back here.
                 "network_policy": properties.get("network_policy"),
                 "pre_authorized_roles_list": pre_authorized_roles_list,
                 "blocked_roles_list": sorted(blocked_roles_list) or None,
                 "comment": data["comment"] or None,
                 "owner": owner,
             }
-    raise Exception(f"Unsupported security integration type {data['type']}")
+        elif oauth_client in ("LOOKER", "TABLEAU_DESKTOP", "TABLEAU_SERVER"):
+            # Partner OAuth is modeled (SnowflakePartnerOAuthSecurityIntegration) and
+            # declarable, but has no fetch branch. Returning None would make a declared
+            # partner integration look absent and plan a spurious CREATE on every apply, so
+            # fail loudly: a declared resource that genuinely can't be read back is an error.
+            raise Exception(
+                f"snowcap cannot read back partner OAuth integration {fqn.name!r} "
+                f"(oauth_client={oauth_client}): fetch is not implemented for partner OAuth"
+            )
+    # A security integration type snowcap does not model (e.g. SAML2, SCIM, EXTERNAL_OAUTH)
+    # must not break list/export just because the account holds one. Skip it with a warning
+    # rather than raising, so fetching one unmodeled type doesn't abort the whole run.
+    logger.warning(f"Skipping unsupported security integration type {data['type']!r} for {fqn.name}")
+    return None
 
 
 def fetch_sequence(session: SnowflakeConnection, fqn: FQN):

snowcap/resources/security_integration.py

@@ -50,6 +50,12 @@ class OAuthUseSecondaryRoles(ParseableEnum):
     NONE = "NONE"
 
 
+class OAuthAnyRoleMode(ParseableEnum):
+    DISABLE = "DISABLE"
+    ENABLE = "ENABLE"
+    ENABLE_FOR_PRIVILEGE = "ENABLE_FOR_PRIVILEGE"
+
+
 @dataclass(unsafe_hash=True)
 class _SnowflakePartnerOAuthSecurityIntegration(ResourceSpec):
     name: ResourceName
@@ -377,11 +383,17 @@ class _SnowflakeCustomOAuthSecurityIntegration(ResourceSpec):
         },
     )
     oauth_redirect_uri: str = None
+    oauth_allow_non_tls_redirect_uri: bool = False
     oauth_alternate_redirect_uris: list[str] = field(default=None, metadata={"fetchable": False})
     oauth_issue_refresh_tokens: bool = True
     oauth_refresh_token_validity: int = 7776000
+    oauth_single_use_refresh_tokens_required: bool = False
     oauth_use_secondary_roles: OAuthUseSecondaryRoles = OAuthUseSecondaryRoles.NONE
+    oauth_any_role_mode: OAuthAnyRoleMode = OAuthAnyRoleMode.DISABLE
     oauth_enforce_pkce: bool = False
+    # CREATE-only: Snowflake accepts it at creation but DESC never echoes it back, so it
+    # can't round-trip. Marking it unfetchable keeps it out of the diff (no phantom drift).
+    oauth_enable_role_selection: bool = field(default=False, metadata={"fetchable": False})
     network_policy: str = None
     pre_authorized_roles_list: list[str] = None
     blocked_roles_list: list[str] = None
@@ -437,11 +449,15 @@ class SnowflakeCustomOAuthSecurityIntegration(NamedResource, Resource):
         enabled (bool): Specifies if the security integration is enabled. Defaults to True.
         oauth_client_type (string or OAuthClientType, required): The type of OAuth client. Supported values are 'CONFIDENTIAL' and 'PUBLIC'. Cannot be changed after creation.
         oauth_redirect_uri (string, required): The redirect URI the client uses to complete the OAuth flow.
+        oauth_allow_non_tls_redirect_uri (bool): Allows non-HTTPS redirect URIs. Defaults to False.
         oauth_alternate_redirect_uris (list): Additional allowed redirect URIs, set at creation only.
         oauth_issue_refresh_tokens (bool): Indicates if refresh tokens should be issued. Defaults to True.
         oauth_refresh_token_validity (int): The validity period of the refresh token in seconds. Defaults to 7776000.
+        oauth_single_use_refresh_tokens_required (bool): Requires refresh tokens to be single-use (rotated on each use). Defaults to False.
         oauth_use_secondary_roles (string or OAuthUseSecondaryRoles): Whether secondary roles are activated for OAuth sessions. Supported values are 'IMPLICIT' and 'NONE'. Defaults to 'NONE'.
+        oauth_any_role_mode (string or OAuthAnyRoleMode): Whether a client can request any role the user has. Supported values are 'DISABLE', 'ENABLE', and 'ENABLE_FOR_PRIVILEGE'. Defaults to 'DISABLE'.
         oauth_enforce_pkce (bool): Requires clients to use PKCE during the OAuth flow. Defaults to False.
+        oauth_enable_role_selection (bool): Lets the client request a role at authorization time. Defaults to False. Set at creation only (DESC does not return it, so it is not reconciled).
         network_policy (string): The network policy enforced for requests made with this integration's tokens.
         pre_authorized_roles_list (list): Roles granted access without displaying a consent screen to the user.
         blocked_roles_list (list): Roles that are not allowed to use this integration.
@@ -491,13 +507,20 @@ class SnowflakeCustomOAuthSecurityIntegration(NamedResource, Resource):
             "oauth_client_type", [OAuthClientType.CONFIDENTIAL, OAuthClientType.PUBLIC], quoted=True
         ),
         oauth_redirect_uri=StringProp("oauth_redirect_uri"),
+        oauth_allow_non_tls_redirect_uri=BoolProp("oauth_allow_non_tls_redirect_uri"),
         oauth_alternate_redirect_uris=StringListProp("oauth_alternate_redirect_uris", parens=True),
         oauth_issue_refresh_tokens=BoolProp("oauth_issue_refresh_tokens"),
         oauth_refresh_token_validity=IntProp("oauth_refresh_token_validity"),
+        oauth_single_use_refresh_tokens_required=BoolProp("oauth_single_use_refresh_tokens_required"),
         oauth_use_secondary_roles=EnumProp(
             "oauth_use_secondary_roles", [OAuthUseSecondaryRoles.IMPLICIT, OAuthUseSecondaryRoles.NONE]
         ),
+        oauth_any_role_mode=EnumProp(
+            "oauth_any_role_mode",
+            [OAuthAnyRoleMode.DISABLE, OAuthAnyRoleMode.ENABLE, OAuthAnyRoleMode.ENABLE_FOR_PRIVILEGE],
+        ),
         oauth_enforce_pkce=BoolProp("oauth_enforce_pkce"),
+        oauth_enable_role_selection=BoolProp("oauth_enable_role_selection"),
         network_policy=StringProp("network_policy"),
         pre_authorized_roles_list=StringListProp("pre_authorized_roles_list", parens=True),
         blocked_roles_list=StringListProp("blocked_roles_list", parens=True),
@@ -512,11 +535,15 @@ def __init__(
         enabled: bool = True,
         oauth_client_type: OAuthClientType = None,
         oauth_redirect_uri: str = None,
+        oauth_allow_non_tls_redirect_uri: bool = False,
         oauth_alternate_redirect_uris: list[str] = None,
         oauth_issue_refresh_tokens: bool = True,
         oauth_refresh_token_validity: int = 7776000,
+        oauth_single_use_refresh_tokens_required: bool = False,
         oauth_use_secondary_roles: OAuthUseSecondaryRoles = OAuthUseSecondaryRoles.NONE,
+        oauth_any_role_mode: OAuthAnyRoleMode = OAuthAnyRoleMode.DISABLE,
         oauth_enforce_pkce: bool = False,
+        oauth_enable_role_selection: bool = False,
         network_policy: str = None,
         pre_authorized_roles_list: list[str] = None,
         blocked_roles_list: list[str] = None,
@@ -532,11 +559,15 @@ def __init__(
             enabled=enabled,
             oauth_client_type=oauth_client_type,
             oauth_redirect_uri=oauth_redirect_uri,
+            oauth_allow_non_tls_redirect_uri=oauth_allow_non_tls_redirect_uri,
             oauth_alternate_redirect_uris=oauth_alternate_redirect_uris,
             oauth_issue_refresh_tokens=oauth_issue_refresh_tokens,
             oauth_refresh_token_validity=oauth_refresh_token_validity,
+            oauth_single_use_refresh_tokens_required=oauth_single_use_refresh_tokens_required,
             oauth_use_secondary_roles=oauth_use_secondary_roles,
+            oauth_any_role_mode=oauth_any_role_mode,
             oauth_enforce_pkce=oauth_enforce_pkce,
+            oauth_enable_role_selection=oauth_enable_role_selection,
             network_policy=network_policy,
             pre_authorized_roles_list=pre_authorized_roles_list,
             blocked_roles_list=blocked_roles_list,

tests/fixtures/json/snowflake_custom_oauth_security_integration.json

@@ -5,11 +5,15 @@
     "oauth_client": "CUSTOM",
     "oauth_client_type": "CONFIDENTIAL",
     "oauth_redirect_uri": "https://example.com/oauth/callback",
+    "oauth_allow_non_tls_redirect_uri": false,
     "oauth_alternate_redirect_uris": ["https://example.com/oauth/callback2"],
     "oauth_issue_refresh_tokens": true,
     "oauth_refresh_token_validity": 7776000,
+    "oauth_single_use_refresh_tokens_required": false,
     "oauth_use_secondary_roles": "NONE",
+    "oauth_any_role_mode": "DISABLE",
     "oauth_enforce_pkce": true,
+    "oauth_enable_role_selection": false,
     "network_policy": null,
     "pre_authorized_roles_list": null,
     "blocked_roles_list": null,

tests/integration/data_provider/test_fetch_resource.py

@@ -417,10 +417,14 @@ def test_fetch_snowflake_custom_oauth_security_integration(cursor, suffix, marke
         name=f"CUSTOM_OAUTH_SECURITY_INTEGRATION_{suffix}",
         oauth_client_type="CONFIDENTIAL",
         oauth_redirect_uri="https://example.com/oauth/callback",
+        oauth_allow_non_tls_redirect_uri=False,
         oauth_enforce_pkce=True,
+        oauth_enable_role_selection=True,
         oauth_issue_refresh_tokens=True,
         oauth_refresh_token_validity=86400,
+        oauth_single_use_refresh_tokens_required=False,
         oauth_use_secondary_roles="IMPLICIT",
+        oauth_any_role_mode="ENABLE_FOR_PRIVILEGE",
         pre_authorized_roles_list=["PUBLIC"],
         comment="Test custom OAuth security integration",
         enabled=True,

tests/test_data_provider.py

@@ -1236,9 +1236,12 @@ def row(property, value, property_type):
     return [
         row("OAUTH_CLIENT_TYPE", "CONFIDENTIAL", "String"),
         row("OAUTH_REDIRECT_URI", "https://example.com/callback", "String"),
+        row("OAUTH_ALLOW_NON_TLS_REDIRECT_URI", "false", "Boolean"),
         row("OAUTH_ISSUE_REFRESH_TOKENS", "true", "Boolean"),
         row("OAUTH_REFRESH_TOKEN_VALIDITY", "7776000", "Long"),
+        row("OAUTH_SINGLE_USE_REFRESH_TOKENS_REQUIRED", "false", "Boolean"),
         row("OAUTH_USE_SECONDARY_ROLES", "NONE", "String"),
+        row("OAUTH_ANY_ROLE_MODE", "DISABLE", "String"),
         row("OAUTH_ENFORCE_PKCE", "false", "Boolean"),
         row("NETWORK_POLICY", "", "String"),
         row("PRE_AUTHORIZED_ROLES_LIST", pre_authorized_roles_list, "List"),
@@ -1375,14 +1378,48 @@ def test_snowservices_ingress_still_fetches_as_before(self, mock_execute, mock_f
 
     @patch("snowcap.data_provider._fetch_owner")
     @patch("snowcap.data_provider.execute")
-    def test_unsupported_oauth_client_raises(self, mock_execute, mock_fetch_owner):
+    def test_custom_oauth_new_fields_round_trip(self, mock_execute, mock_fetch_owner):
+        """The serverless-OAuth toggles read back from DESC so they don't drift on every plan."""
+        mock_fetch_owner.return_value = "ACCOUNTADMIN"
+        mock_execute.side_effect = self._mock_execute(_custom_oauth_desc_rows())
+
+        result = fetch_security_integration(MagicMock(), FQN(name=ResourceName("CUSTOM_OAUTH")))
+
+        assert result["oauth_allow_non_tls_redirect_uri"] is False
+        assert result["oauth_single_use_refresh_tokens_required"] is False
+        assert result["oauth_any_role_mode"] == "DISABLE"
+        # oauth_enable_role_selection is CREATE-only (not in DESC), so fetch omits it.
+        assert "oauth_enable_role_selection" not in result
+
+    @patch("snowcap.data_provider._fetch_owner")
+    @patch("snowcap.data_provider.execute")
+    def test_unmodeled_type_returns_none_with_warning(self, mock_execute, mock_fetch_owner, caplog):
+        """A type snowcap doesn't model (e.g. SAML2) must not raise: it would break list/export
+        whenever the account holds one. Return None and warn instead."""
+        mock_fetch_owner.return_value = "ACCOUNTADMIN"
+        mock_execute.side_effect = self._mock_execute(
+            desc_rows=[],
+            show_row=_security_integration_show_row(name="MY_SAML", type="SAML2"),
+        )
+
+        with caplog.at_level("WARNING", logger="snowcap"):
+            result = fetch_security_integration(MagicMock(), FQN(name=ResourceName("MY_SAML")))
+
+        assert result is None
+        assert "unsupported security integration type" in caplog.text.lower()
+
+    @patch("snowcap.data_provider._fetch_owner")
+    @patch("snowcap.data_provider.execute")
+    def test_partner_oauth_still_raises(self, mock_execute, mock_fetch_owner):
+        """Partner OAuth is modeled/declarable but not fetchable, so it must still raise:
+        returning None would make a declared partner integration look absent and churn a CREATE."""
         mock_fetch_owner.return_value = "ACCOUNTADMIN"
         mock_execute.side_effect = self._mock_execute(
             desc_rows=[],
             show_row=_security_integration_show_row(name="TABLEAU", type="OAUTH - TABLEAU_DESKTOP"),
         )
 
-        with pytest.raises(Exception, match="Unsupported security integration type"):
+        with pytest.raises(Exception, match="cannot read back partner OAuth"):
             fetch_security_integration(MagicMock(), FQN(name=ResourceName("TABLEAU")))
 
 

tests/test_lifecycle.py

@@ -1649,9 +1649,12 @@ def test_create_sql(self):
             "CREATE SECURITY INTEGRATION CLAUDE_MCP_OAUTH type = OAUTH ENABLED = TRUE "
             "oauth_client = CUSTOM oauth_client_type = 'CONFIDENTIAL' "
             "OAUTH_REDIRECT_URI = $$https://example.com/cb$$ "
+            "OAUTH_ALLOW_NON_TLS_REDIRECT_URI = FALSE "
             "OAUTH_ALTERNATE_REDIRECT_URIS = ($$https://example.com/cb2$$) "
             "OAUTH_ISSUE_REFRESH_TOKENS = TRUE OAUTH_REFRESH_TOKEN_VALIDITY = 7776000 "
-            "oauth_use_secondary_roles = NONE OAUTH_ENFORCE_PKCE = TRUE "
+            "OAUTH_SINGLE_USE_REFRESH_TOKENS_REQUIRED = FALSE "
+            "oauth_use_secondary_roles = NONE oauth_any_role_mode = DISABLE "
+            "OAUTH_ENFORCE_PKCE = TRUE OAUTH_ENABLE_ROLE_SELECTION = FALSE "
             "BLOCKED_ROLES_LIST = ($$ANALYST$$, $$SYSADMIN$$) COMMENT = $$test comment$$"
         )
 
@@ -1667,6 +1670,9 @@ def test_metadata(self):
         # The plan-time error must explain why snowcap refuses to recreate.
         assert "client_id and client_secret" in oauth_client_type_metadata.replacement_message
         assert spec.get_metadata("oauth_alternate_redirect_uris").fetchable is False
+        # oauth_enable_role_selection is CREATE-only: DESC never returns it, so it must be
+        # unfetchable or it drifts "set to default" on every plan.
+        assert spec.get_metadata("oauth_enable_role_selection").fetchable is False
 
     def test_update_set_sql(self):
         """A single-field delta renders as ALTER SECURITY INTEGRATION ... SET, never CREATE OR REPLACE."""

tests/test_polymorphic_resources.py

@@ -44,6 +44,13 @@ def test_view_stream():
     assert isinstance(resource, res.ViewStream)
 
 
+def test_custom_oauth_security_integration():
+    resource_cls = Resource.resolve_resource_cls(
+        ResourceType.SECURITY_INTEGRATION, {"type": "OAUTH", "oauth_client": "CUSTOM"}
+    )
+    assert resource_cls is res.SnowflakeCustomOAuthSecurityIntegration
+
+
 def enumerate_polymorphic_resources():
     """Get polymorphic resources that have resolvers (can be distinguished by data)."""
     # List of resource fixtures that have been intentionally removed because they

tests/test_yaml_config.py

@@ -692,3 +692,37 @@ def test_imported_privileges_grant_create_sql(self):
         grant = blueprint_config.resources[0]
         assert isinstance(grant, res.Grant)
         assert grant.create_sql() == "GRANT IMPORTED PRIVILEGES ON DATABASE GONG TO ROLE GONG_R"
+
+
+class TestSecurityIntegrationConfig:
+    """A security_integrations: block with oauth_client: CUSTOM resolves to the custom OAuth resource."""
+
+    def test_custom_oauth_from_yaml(self):
+        from snowcap import resources as res
+
+        config = {
+            "security_integrations": [
+                {
+                    "name": "claude_mcp_oauth",
+                    "type": "OAUTH",
+                    "oauth_client": "CUSTOM",
+                    "oauth_client_type": "CONFIDENTIAL",
+                    "oauth_redirect_uri": "https://claude.ai/api/mcp/auth_callback",
+                    "oauth_enforce_pkce": True,
+                    "oauth_any_role_mode": "ENABLE_FOR_PRIVILEGE",
+                    "oauth_allow_non_tls_redirect_uri": False,
+                    "blocked_roles_list": ["SYSADMIN"],
+                    "comment": "OAuth client for the Claude MCP connector",
+                }
+            ],
+        }
+        blueprint_config = collect_blueprint_config(config)
+
+        assert len(blueprint_config.resources) == 1
+        integration = blueprint_config.resources[0]
+        assert isinstance(integration, res.SnowflakeCustomOAuthSecurityIntegration)
+        data = integration.to_dict()
+        assert data["oauth_client_type"] == "CONFIDENTIAL"
+        assert data["oauth_any_role_mode"] == "ENABLE_FOR_PRIVILEGE"
+        assert data["oauth_enforce_pkce"] is True
+        assert data["blocked_roles_list"] == ["SYSADMIN"]

version.md

@@ -1 +1 @@
-# version 1.0.23
+# version 1.0.24