From 0984285fcce025769e26a2576a4231da922e64a6 Mon Sep 17 00:00:00 2001 From: Noel Gomez Date: Wed, 9 Sep 2026 07:46:14 -0700 Subject: [PATCH 1/2] fix(grant): stop phantom CREATE for database-role-to-database-role grants Snowflake reports a DATABASE_ROLE grantee unqualified when it's in the same database as the granting role, but the manifest side always compares against a fully qualified DB.ROLE string. That mismatch made fetch_database_role_grant and list_database_role_grants never find the existing grant, so plan re-issued it as a no-op CREATE on every run. - fetch_database_role_grant and both list_database_role_grants paths now qualify-then-compare grantee names via a shared _database_role_grantee_fqn() helper instead of comparing raw strings. - _fetch_grants_from_account_usage now normalizes GRANTED_ON the same way it already normalizes GRANTED_TO (DATABASE_ROLE -> DATABASE ROLE), fixing a dead filter that made the ACCOUNT_USAGE path for list_database_role_grants always return zero results. --- snowcap/data_provider.py | 48 ++++++- tests/test_data_provider.py | 245 ++++++++++++++++++++++++++++++++++++ 2 files changed, 288 insertions(+), 5 deletions(-) diff --git a/snowcap/data_provider.py b/snowcap/data_provider.py index edf170b..47cc3c6 100644 --- a/snowcap/data_provider.py +++ b/snowcap/data_provider.py @@ -1524,7 +1524,10 @@ def _fetch_grants_from_account_usage(session: SnowflakeConnection) -> list[dict[ # Construct fully qualified name to match SHOW GRANTS output # ACCOUNT_USAGE NAME column only has object name, not full path + # Same 'DATABASE_ROLE' -> 'DATABASE ROLE' normalization as granted_to above granted_on = row["GRANTED_ON"] + if granted_on == "DATABASE_ROLE": + granted_on = "DATABASE ROLE" name = row["NAME"] table_catalog = row.get("TABLE_CATALOG") table_schema = row.get("TABLE_SCHEMA") @@ -1539,7 +1542,7 @@ def _fetch_grants_from_account_usage(session: SnowflakeConnection) -> list[dict[ # Schema grants: need DATABASE.SCHEMA if table_catalog: name = f"{table_catalog}.{name}" - elif granted_on in ("DATABASE ROLE",): + elif granted_on == "DATABASE ROLE": # Database role grants: need DATABASE.ROLE if table_catalog: name = f"{table_catalog}.{name}" @@ -2106,12 +2109,37 @@ def fetch_database_role(session: SnowflakeConnection, fqn: FQN): } +def _database_role_grantee_fqn(grantee_name: str, default_database: ResourceName) -> FQN: + """ + Build an FQN for a DATABASE_ROLE grant's grantee. + + Snowflake reports a database-role grantee unqualified when it lives in the same + database as the role granting it, but fully qualified when it's in another database. + Parsing it as an FQN and filling in default_database only when it's missing makes both + cases comparable to DatabaseRoleGrant.to_database_role's FQN, which is always fully + qualified. + """ + grantee_fqn = parse_FQN(grantee_name, is_db_scoped=True) + if grantee_fqn.database is None: + grantee_fqn.database = default_database + return grantee_fqn + + def fetch_database_role_grant(session: SnowflakeConnection, fqn: FQN): show_result = execute(session, f"SHOW GRANTS OF DATABASE ROLE {fqn.database}.{fqn.name}", cacheable=True) subject, subject_name = next(iter(fqn.params.items())) - role_grants = _filter_result(show_result, granted_to=subject.upper(), grantee_name=subject_name) + if subject == "database_role": + target = parse_FQN(subject_name, is_db_scoped=True) + role_grants = [ + row + for row in _filter_result(show_result, granted_to=subject.upper()) + if _database_role_grantee_fqn(row["grantee_name"], fqn.database) == target + ] + else: + role_grants = _filter_result(show_result, granted_to=subject.upper(), grantee_name=subject_name) + if len(role_grants) == 0: return None if len(role_grants) > 1: @@ -2124,7 +2152,7 @@ def fetch_database_role_grant(session: SnowflakeConnection, fqn: FQN): if data["granted_to"] == "ROLE": to_role = _quote_snowflake_identifier(data["grantee_name"]) elif data["granted_to"] == "DATABASE_ROLE": - to_database_role = data["grantee_name"] + to_database_role = str(_database_role_grantee_fqn(data["grantee_name"], fqn.database)) return { "database_role": data["role"], @@ -4279,12 +4307,17 @@ def list_database_role_grants( # Determine subject based on grantee type subject = "role" if grant["granted_to"] == "ROLE" else "database_role" + grantee_name = grant["grantee_name"] + if subject == "database_role": + grantee_name = str( + _database_role_grantee_fqn(grantee_name, resource_name_from_snowflake_metadata(db_name)) + ) role_grants.append( FQN( name=resource_name_from_snowflake_metadata(role_name), database=resource_name_from_snowflake_metadata(db_name), - params={subject: grant["grantee_name"]}, + params={subject: grantee_name}, ) ) # If we got results or no specific database was filtered, return @@ -4318,11 +4351,16 @@ def list_database_role_grants( for data in show_result: subject = "role" if data["granted_to"] == "ROLE" else "database_role" db, name = data["role"].split(".") + grantee_name = data["grantee_name"] + if subject == "database_role": + grantee_name = str( + _database_role_grantee_fqn(grantee_name, resource_name_from_snowflake_metadata(db)) + ) role_grants.append( FQN( name=resource_name_from_snowflake_metadata(name), database=resource_name_from_snowflake_metadata(db), - params={subject: data["grantee_name"]}, + params={subject: grantee_name}, ) ) return role_grants diff --git a/tests/test_data_provider.py b/tests/test_data_provider.py index 34c4d8d..5433821 100644 --- a/tests/test_data_provider.py +++ b/tests/test_data_provider.py @@ -2490,6 +2490,251 @@ def execute_side_effect(session, query, **kwargs): assert not [on for on in on_values if "database_role" in on] +class TestDatabaseRoleGrantToDatabaseRole: + """Snowflake reports a DATABASE_ROLE grantee unqualified when it's in the same database + as the role granting it, but the manifest side always builds a fully qualified + DB.ROLE string. Comparing the two directly never matches, so plan re-issues the grant + as a no-op CREATE on every run.""" + + @patch("snowcap.data_provider.execute") + def test_fetch_matches_unqualified_grantee_in_same_database(self, mock_execute): + from snowcap.data_provider import fetch_database_role_grant + from snowcap.identifiers import FQN + from snowcap.resource_name import ResourceName + + mock_execute.return_value = [ + { + "role": "GREAT_BAY_DEV.DR_READER_ROLE", + "granted_to": "DATABASE_ROLE", + "grantee_name": "DR_WRITER_ROLE", + "granted_by": "SECURITYADMIN", + } + ] + fqn = FQN( + name=ResourceName("DR_READER_ROLE"), + database=ResourceName("GREAT_BAY_DEV"), + params={"database_role": "GREAT_BAY_DEV.DR_WRITER_ROLE"}, + ) + + result = fetch_database_role_grant(MagicMock(), fqn) + + assert result is not None + assert result["to_database_role"] == "GREAT_BAY_DEV.DR_WRITER_ROLE" + + @patch("snowcap.data_provider.execute") + def test_fetch_matches_qualified_grantee_in_another_database(self, mock_execute): + from snowcap.data_provider import fetch_database_role_grant + from snowcap.identifiers import FQN + from snowcap.resource_name import ResourceName + + mock_execute.return_value = [ + { + "role": "GREAT_BAY_DEV.DR_READER_ROLE", + "granted_to": "DATABASE_ROLE", + "grantee_name": "OTHER_DB.DR_WRITER_ROLE", + "granted_by": "SECURITYADMIN", + } + ] + fqn = FQN( + name=ResourceName("DR_READER_ROLE"), + database=ResourceName("GREAT_BAY_DEV"), + params={"database_role": "OTHER_DB.DR_WRITER_ROLE"}, + ) + + result = fetch_database_role_grant(MagicMock(), fqn) + + assert result is not None + assert result["to_database_role"] == "OTHER_DB.DR_WRITER_ROLE" + + @patch("snowcap.data_provider.execute") + def test_fetch_does_not_false_match_grantee_in_a_different_database(self, mock_execute): + """OTHERDB.DR_WRITER_ROLE must not match a target of DB.DR_WRITER_ROLE just because + the bare role name is the same.""" + from snowcap.data_provider import fetch_database_role_grant + from snowcap.identifiers import FQN + from snowcap.resource_name import ResourceName + + mock_execute.return_value = [ + { + "role": "GREAT_BAY_DEV.DR_READER_ROLE", + "granted_to": "DATABASE_ROLE", + "grantee_name": "OTHER_DB.DR_WRITER_ROLE", + "granted_by": "SECURITYADMIN", + } + ] + fqn = FQN( + name=ResourceName("DR_READER_ROLE"), + database=ResourceName("GREAT_BAY_DEV"), + params={"database_role": "GREAT_BAY_DEV.DR_WRITER_ROLE"}, + ) + + assert fetch_database_role_grant(MagicMock(), fqn) is None + + @patch("snowcap.data_provider.execute") + def test_fetch_regression_grant_to_account_role(self, mock_execute): + from snowcap.data_provider import fetch_database_role_grant + from snowcap.identifiers import FQN + from snowcap.resource_name import ResourceName + + mock_execute.return_value = [ + { + "role": "GREAT_BAY_DEV.DR_READER_ROLE", + "granted_to": "ROLE", + "grantee_name": "GREAT_BAY_DEV__READER", + "granted_by": "SECURITYADMIN", + } + ] + fqn = FQN( + name=ResourceName("DR_READER_ROLE"), + database=ResourceName("GREAT_BAY_DEV"), + params={"role": "GREAT_BAY_DEV__READER"}, + ) + + result = fetch_database_role_grant(MagicMock(), fqn) + + assert result is not None + assert result["to_role"] == "GREAT_BAY_DEV__READER" + + @patch("snowcap.data_provider._should_use_account_usage") + @patch("snowcap.data_provider.execute") + def test_list_show_path_qualifies_same_database_grantee(self, mock_execute, mock_should_use): + from snowcap.data_provider import list_database_role_grants + + mock_should_use.return_value = False + + def execute_side_effect(session, query, **kwargs): + if "SHOW DATABASE ROLES IN DATABASE" in query: + return [{"name": "DR_READER_ROLE"}] + if "SHOW GRANTS OF DATABASE ROLE" in query: + return [ + { + "role": "GREAT_BAY_DEV.DR_READER_ROLE", + "granted_to": "DATABASE_ROLE", + "grantee_name": "DR_WRITER_ROLE", + "granted_by": "SECURITYADMIN", + } + ] + raise AssertionError(f"unexpected query: {query}") + + mock_execute.side_effect = execute_side_effect + + grants = list_database_role_grants(MagicMock(), database="GREAT_BAY_DEV") + + assert len(grants) == 1 + assert grants[0].params == {"database_role": "GREAT_BAY_DEV.DR_WRITER_ROLE"} + + @patch("snowcap.data_provider._should_use_account_usage") + @patch("snowcap.data_provider.execute") + def test_list_account_usage_path_recognizes_underscore_granted_on(self, mock_execute, mock_should_use): + """SNOWFLAKE.ACCOUNT_USAGE.GRANTS_TO_ROLES spells the object type GRANTED_ON = + 'DATABASE_ROLE' (underscore). _fetch_grants_from_account_usage must normalize that + to 'DATABASE ROLE' (matching SHOW GRANTS) or list_database_role_grants's filter on + granted_on never matches and this path always returns zero results.""" + from datetime import datetime + + from snowcap.data_provider import list_database_role_grants + + mock_should_use.return_value = True + mock_execute.return_value = [ + { + "CREATED_ON": datetime(2024, 1, 1), + "PRIVILEGE": "USAGE", + "GRANTED_ON": "DATABASE_ROLE", + "NAME": "DR_READER_ROLE", + "TABLE_CATALOG": "GREAT_BAY_DEV", + "GRANTED_TO": "DATABASE_ROLE", + "GRANTEE_NAME": "DR_WRITER_ROLE", + "GRANT_OPTION": False, + "GRANTED_BY": "SECURITYADMIN", + } + ] + + grants = list_database_role_grants(MagicMock(), database="GREAT_BAY_DEV", use_account_usage=True) + + assert len(grants) == 1 + assert grants[0].params == {"database_role": "GREAT_BAY_DEV.DR_WRITER_ROLE"} + + def test_account_usage_and_show_paths_return_identical_fqns(self): + """The two list paths are interchangeable sources for the same diff, so they must + agree on the FQN they produce for the same underlying grant.""" + from snowcap.data_provider import list_database_role_grants + + au_row = { + "privilege": "USAGE", + "granted_on": "DATABASE ROLE", # as normalized by _fetch_grants_from_account_usage + "name": "GREAT_BAY_DEV.DR_READER_ROLE", + "granted_to": "DATABASE_ROLE", + "grantee_name": "DR_WRITER_ROLE", + } + show_row = { + "role": "GREAT_BAY_DEV.DR_READER_ROLE", + "granted_to": "DATABASE_ROLE", + "grantee_name": "DR_WRITER_ROLE", + "granted_by": "SECURITYADMIN", + } + + with ( + patch("snowcap.data_provider._should_use_account_usage", return_value=True), + patch("snowcap.data_provider._fetch_grants_from_account_usage", return_value=[au_row]), + ): + au_grants = list_database_role_grants(MagicMock(), database="GREAT_BAY_DEV", use_account_usage=True) + + def show_side_effect(session, query, **kwargs): + if "SHOW DATABASE ROLES IN DATABASE" in query: + return [{"name": "DR_READER_ROLE"}] + if "SHOW GRANTS OF DATABASE ROLE" in query: + return [show_row] + raise AssertionError(f"unexpected query: {query}") + + with ( + patch("snowcap.data_provider._should_use_account_usage", return_value=False), + patch("snowcap.data_provider.execute", side_effect=show_side_effect), + ): + show_grants = list_database_role_grants(MagicMock(), database="GREAT_BAY_DEV") + + assert len(au_grants) == 1 + assert len(show_grants) == 1 + assert au_grants[0] == show_grants[0] + assert au_grants[0].params == {"database_role": "GREAT_BAY_DEV.DR_WRITER_ROLE"} + + def test_fetched_grant_matches_declared_manifest_fqn(self): + """The actual reported symptom: plan compares the fetched FQN against the FQN the + manifest builds for the same declared grant. They must be equal, or plan proposes a + CREATE for a grant that already exists.""" + from snowcap import resources as res + from snowcap.resources.grant import database_role_grant_fqn + from snowcap.data_provider import list_database_role_grants + + grant = res.DatabaseRoleGrant( + database_role="great_bay_dev.dr_reader_role", + to_database_role="great_bay_dev.dr_writer_role", + ) + declared_fqn = database_role_grant_fqn(grant._data) + + def show_side_effect(session, query, **kwargs): + if "SHOW DATABASE ROLES IN DATABASE" in query: + return [{"name": "DR_READER_ROLE"}] + if "SHOW GRANTS OF DATABASE ROLE" in query: + return [ + { + "role": "GREAT_BAY_DEV.DR_READER_ROLE", + "granted_to": "DATABASE_ROLE", + "grantee_name": "DR_WRITER_ROLE", + "granted_by": "SECURITYADMIN", + } + ] + raise AssertionError(f"unexpected query: {query}") + + with ( + patch("snowcap.data_provider._should_use_account_usage", return_value=False), + patch("snowcap.data_provider.execute", side_effect=show_side_effect), + ): + fetched_grants = list_database_role_grants(MagicMock(), database="GREAT_BAY_DEV") + + assert len(fetched_grants) == 1 + assert fetched_grants[0] == declared_fqn + + class TestGrantsReportedUnderASynonym: """Snowflake reports a grant on an MCP server as CORTEX_AGENT_SERVER, while GRANT and CREATE call the object an MCP SERVER. Remote state and the manifest have to identify it From 08f729c12f22f36a3c8d4f915a2aba5599f12278 Mon Sep 17 00:00:00 2001 From: Noel Gomez Date: Wed, 9 Sep 2026 07:58:20 -0700 Subject: [PATCH 2/2] fix(grant): satisfy mypy on _database_role_grantee_fqn's default_database param FQN.database is Optional[ResourceName]; the helper's parameter was typed as the non-optional ResourceName, which mypy flagged at both call sites in fetch_database_role_grant where fqn.database is passed straight through. --- snowcap/data_provider.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/snowcap/data_provider.py b/snowcap/data_provider.py index 47cc3c6..3137b2f 100644 --- a/snowcap/data_provider.py +++ b/snowcap/data_provider.py @@ -2109,7 +2109,7 @@ def fetch_database_role(session: SnowflakeConnection, fqn: FQN): } -def _database_role_grantee_fqn(grantee_name: str, default_database: ResourceName) -> FQN: +def _database_role_grantee_fqn(grantee_name: str, default_database: Optional[ResourceName]) -> FQN: """ Build an FQN for a DATABASE_ROLE grant's grantee.