policyweaver.plugins.snowflake¶

policyweaver.plugins.snowflake.api¶

class policyweaver.plugins.snowflake.api.SnowflakeAPIClient¶

Bases: object

Snowflake API Client for fetching account policies. This client uses the Snowflake SDK to interact with the Snowflake account and retrieve users, databases, schemas, tables, and privileges. This class is designed to be used within the Policy Weaver framework to gather and map policies from Snowflake workspaces and accounts.

policyweaver.plugins.snowflake.client¶

class policyweaver.plugins.snowflake.client.SnowflakePolicyWeaver(config: SnowflakeSourceMap)¶

Bases: PolicyWeaverCore

Snowflake Policy Weaver for Snowflake Databases. This class extends the PolicyWeaverCore to implement the mapping of policies from Snowflake Database to the Policy Weaver framework.

__config_validation(config: SnowflakeSourceMap) → None¶

Validates the configuration for the SnowflakePolicyWeaver. This method checks if the configuration is of type SnowflakeSourceMap and if all required fields are present. Args:

config (SnowflakeSourceMap): The configuration object to validate.

Raises:: ValueError: If the configuration is not of type SnowflakeSourceMap or if any required fields are missing.

__init_environment(config: SnowflakeSourceMap) → None¶

_build_role_based_policy__(grantee_name: str, grants: list[SnowflakeGrant], column_security: bool, row_security: bool) → RolePolicy¶

_build_table_based_policy__(table_catalog: str, table_schema: str, table_name: str, grants: List[SnowflakeGrant]) → Policy¶

map_policy(policy_mapping='role_based')¶: Map policies from the configured source to a unified format. This method retrieves policies from the source, processes them, and returns a PolicyExport object containing the mapped policies. Returns:

PolicyExport: An object containing the mapped policies.

sf_database_read_prereqs = ['USAGE', 'OWNERSHIP']¶

sf_read_permissions = ['SELECT', 'OWNERSHIP']¶

sf_schema_read_prereqs = ['USAGE', 'OWNERSHIP']¶

policyweaver.plugins.snowflake.model¶

class policyweaver.plugins.snowflake.model.RowFilterDetailGroup(*, group_name: str | None = None, return_value: str | None = None)¶

Bases: CommonBaseModel

Represents a group within a row filter, including its name and return value. Attributes:

group_name (Optional[str]): The name of the group associated with the row filter. return_value (Optional[str]): The value returned by the row filter for this group.

_abc_impl = <_abc._abc_data object>¶

group_name: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

return_value: str | None¶

class policyweaver.plugins.snowflake.model.RowFilterDetails(*, groups: List[RowFilterDetailGroup] | None = None, row_filter_type: RowFilterType | None = None, default_value: str | None = None)¶

Bases: CommonBaseModel

Represents the extraction of a row filter from a SQL definition. This class is used to capture the details of a row filter, including the group name and condition for others. Attributes:

group_name (Optional[str]): The name of the group associated with the row filter. row_filter_type (Optional[RowFilterType]): The type of the row filter (e.g., EXPLICIT_GROUP_MEMBERSHIP). default_value (Optional[str]): The default value returned by the row filter when no group matches.

_abc_impl = <_abc._abc_data object>¶

default_value: str | None¶

groups: List[RowFilterDetailGroup] | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

row_filter_type: RowFilterType | None¶

Bases: CommonBaseModel

Represents a column mask that can be applied to data in the Databricks workspace. This class extends BaseObject to include the routine definition of the mask. Attributes:

name: (Optional[str]): The name of the column mask . routine_definition (Optional[str]): The SQL definition of the column mask routine. column_name (Optional[str]): The name of the column to which the mask applies.

_abc_impl = <_abc._abc_data object>¶

column_name: str | None¶

group_names: List[str] | None¶

mask_pattern: str | None¶

mask_type: ColumnMaskType | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None¶

routine_definition: str | None¶

class policyweaver.plugins.snowflake.model.SnowflakeColumnMaskExtraction(*, group_name: List[str] | None = None, mask_pattern: str | None = None, column_mask_type: ColumnMaskType | None = None)¶

Bases: CommonBaseModel

Represents the result of extracting group name and mask pattern from a column mask function definition. Attributes:

group_name (Optional[str]): The name of the group associated with the column mask. mask_pattern (Optional[str]): The mask pattern applied by the column mask. column_mask_type (Optional[ColumnMaskType]): The type of the column mask (e.g., UNMASK_FOR_GROUP, MASK_FOR_GROUP, UNSUPPORTED).

_abc_impl = <_abc._abc_data object>¶

column_mask_type: ColumnMaskType | None¶

group_names: List[str] | None¶

mask_pattern: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Bases: CommonBaseModel

Represents a connection to a Snowflake account. Attributes:

account_name (Optional[str]): The name of the Snowflake account. user_name (Optional[str]): The user name for accessing the Snowflake account. password (Optional[str]): The password for accessing the Snowflake account. warehouse (Optional[str]): The warehouse to use for the Snowflake connection.

_abc_impl = <_abc._abc_data object>¶

account_name: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

password: str | None¶

private_key_file: str | None¶

user_name: str | None¶

warehouse: str | None¶

class policyweaver.plugins.snowflake.model.SnowflakeDatabaseMap(*, users: ~typing.List[~policyweaver.plugins.snowflake.model.SnowflakeUser] = <factory>, roles: ~typing.List[~policyweaver.plugins.snowflake.model.SnowflakeRole] = <factory>, grants: ~typing.List[~policyweaver.plugins.snowflake.model.SnowflakeGrant] = <factory>, masking_policies: ~typing.List[~policyweaver.plugins.snowflake.model.SnowflakeMaskingPolicy] = <factory>, tables_with_masks: ~typing.List[~policyweaver.plugins.snowflake.model.SnowflakeTableWithPolicy] = <factory>, row_access_policies: ~typing.List[~policyweaver.plugins.snowflake.model.SnowflakeRowFilter] = <factory>, tables_with_raps: ~typing.List[~policyweaver.plugins.snowflake.model.SnowflakeTableWithPolicy] = <factory>, unsupported_tables: ~typing.List[~policyweaver.plugins.snowflake.model.SnowflakeTableWithPolicy] = <factory>)¶

Bases: CommonBaseModel

A collection of Snowflake users, roles, and grants for a database Attributes:

users (List[SnowflakeUser]): The list of users in the Snowflake database. roles (List[SnowflakeRole]): The list of roles in the Snowflake database. grants (List[SnowflakeGrant]): The list of grants in the Snowflake database. masking_policies (List[SnowflakeMaskingPolicy]): The list of masking policies in the Snowflake database.

_abc_impl = <_abc._abc_data object>¶

grants: List[SnowflakeGrant]¶

masking_policies: List[SnowflakeMaskingPolicy]¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

roles: List[SnowflakeRole]¶

row_access_policies: List[SnowflakeRowFilter]¶

tables_with_masks: List[SnowflakeTableWithPolicy]¶

tables_with_raps: List[SnowflakeTableWithPolicy]¶

unsupported_tables: List[SnowflakeTableWithPolicy]¶

users: List[SnowflakeUser]¶

Bases: CommonBaseModel

Represents a grant in the Snowflake workspace. Attributes:

privilege (Optional[str]): The privilege granted. granted_on (Optional[str]): The object type on which the privilege is granted. table_catalog (Optional[str]): The catalog of the table. table_schema (Optional[str]): The schema of the table. name (Optional[str]): The name of the object. grantee_name (Optional[str]): The name of the grantee (user or role).

_abc_impl = <_abc._abc_data object>¶

granted_on: str | None¶

grantee_name: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None¶

privilege: str | None¶

table_catalog: str | None¶

table_schema: str | None¶

Bases: CommonBaseModel

Represents a masking policy in the Snowflake workspace. Attributes:

id (Optional[int]): The unique identifier for the masking policy. name (Optional[str]): The name of the masking policy. database_name (Optional[str]): The name of the database to which the masking policy applies. schema_name (Optional[str]): The name of the schema to which the masking policy applies. entity_name (Optional[str]): The name of the entity to which the masking policy applies. column_name (Optional[str]): The name of the column to which the masking policy applies.

_abc_impl = <_abc._abc_data object>¶

column_mask_type: ColumnMaskType | None¶

column_name: str | None¶

database_name: str | None¶

group_names: List[str] | None¶

id: int | None¶

mask_pattern: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None¶

schema_name: str | None¶

table_name: str | None¶

class policyweaver.plugins.snowflake.model.SnowflakeRole(*, id: int | None = None, name: str | None = None, members_user: ~typing.List[~policyweaver.plugins.snowflake.model.SnowflakeUser] = <factory>, members_role: ~typing.List[~policyweaver.plugins.snowflake.model.SnowflakeRole] = <factory>, role_assignments: ~typing.List[~policyweaver.plugins.snowflake.model.SnowflakeRole] = <factory>)¶

Bases: SnowflakeUserOrRole

Represents a role in the Snowflake workspace. This class extends BaseObject to include additional attributes specific to roles. Attributes:

id (Optional[int]): The unique identifier for the role. name (Optional[str]): The name of the role. members_user (List[SnowflakeUser]): The users that are assigned to this role. members_role (List[SnowflakeRole]): The roles that are assigned to this role. role_assignments (List[SnowflakeRole]): The roles that this role is assigned to.

_abc_impl = <_abc._abc_data object>¶

id: int | None¶

members_role: List[SnowflakeRole]¶

members_user: List[SnowflakeUser]¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None¶

role_assignments: List[SnowflakeRole]¶

class policyweaver.plugins.snowflake.model.SnowflakeRoleMemberMap(*, role_name: str | None = None, users: ~typing.List[~policyweaver.plugins.snowflake.model.SnowflakeUser] = <factory>, roles: ~typing.List[~policyweaver.plugins.snowflake.model.SnowflakeRole] = <factory>)¶

Bases: CommonBaseModel

Represents the members of a Snowflake role. This class includes the users and roles that are members of the role. Attributes:

users (List[SnowflakeUser]): The users that are members of the role. roles (List[SnowflakeRole]): The roles that are members of the role.

_abc_impl = <_abc._abc_data object>¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

role_name: str | None¶

roles: List[SnowflakeRole]¶

users: List[SnowflakeUser]¶

Bases: CommonBaseModel

Represents a row filter that can be applied to data in the Databricks workspace. This class extends CommonBaseModel to include the routine definition of the filter. Attributes:

name: (Optional[str]): The name of the row filter. routine_definition (Optional[str]): The SQL definition of the row filter routine.

_abc_impl = <_abc._abc_data object>¶

database_name: str | None¶

details: RowFilterDetails | None¶

id: int | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None¶

routine_definition: str | None¶

schema_name: str | None¶

table_name: str | None¶

Bases: CommonBaseModel

Represents the configuration for a Snowflake source. This class includes the account name, user name, and password. Attributes:

account_name (Optional[str]): The name of the Snowflake account. user_name (Optional[str]): The user name for accessing the Snowflake account. password (Optional[str]): The password for accessing the Snowflake account. warehouse (Optional[str]): The warehouse to use for the Snowflake connection. private_key_file (Optional[str]): The path to the private key file for accessing the Snowflake account.

_abc_impl = <_abc._abc_data object>¶

account_name: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

password: str | None¶

private_key_file: str | None¶

user_name: str | None¶

warehouse: str | None¶

class policyweaver.plugins.snowflake.model.SnowflakeSourceMap(*, application_name: str | None = 'POLICY_WEAVER', correlation_id: str | None = None, type: PolicyWeaverConnectorType | None = None, source: Source | None = None, fabric: FabricConfig | None = None, constraints: ConstraintsConfig | None = None, service_principal: ServicePrincipalConfig | None = None, mapped_items: List[SourceMapItem] | None = None, keyvault: KeyVaultConfig | None = None, snowflake: SnowflakeSourceConfig | None = None)¶

Bases: SourceMap

Represents the configuration for a Snowflake source map. This class extends SourceMap to include Snowflake-specific configuration. Attributes:

snowflake (Optional[SnowflakeSourceConfig]): The Snowflake source configuration.

_abc_impl = <_abc._abc_data object>¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None¶

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

snowflake: SnowflakeSourceConfig | None¶

class policyweaver.plugins.snowflake.model.SnowflakeTableWithPolicy(*, database_name: str | None = None, schema_name: str | None = None, table_name: str | None = None, column_names: List[str] | None = None)¶

Bases: CommonBaseModel

Represents a table in Snowflake that has column masks applied. Attributes:

database_name (Optional[str]): The name of the database containing the table. schema_name (Optional[str]): The name of the schema containing the table. table_name (Optional[str]): The name of the table. column_names (Optional[List[str]]): A list of column names in the table that have masks applied.

_abc_impl = <_abc._abc_data object>¶

column_names: List[str] | None¶

database_name: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

schema_name: str | None¶

table_name: str | None¶

class policyweaver.plugins.snowflake.model.SnowflakeUser(*, id: int | None = None, name: str | None = None)¶

Bases: SnowflakeUserOrRole

Represents a user in the Snowflake workspace. This class extends BaseObject to include additional attributes specific to users. Attributes:

id (Optional[int]): The unique identifier for the user. name (Optional[str]): The name of the user. email (Optional[str]): The email address of the user. login_name (Optional[str]): The login name of the user. role_assignments (List[SnowflakeRole]): The roles that this user is assigned to.

_abc_impl = <_abc._abc_data object>¶

email: str | None¶

id: int | None¶

login_name: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None¶

role_assignments: List[SnowflakeRole]¶

class policyweaver.plugins.snowflake.model.SnowflakeUserOrRole(*, id: int | None = None, name: str | None = None)¶

Bases: CommonBaseModel

Represents a user or role in the Snowflake workspace. This class is a base class for both users and roles.

_abc_impl = <_abc._abc_data object>¶

id: int | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None¶