diff --git a/CHANGELOG.md b/CHANGELOG.md index 22650cf3..daf25619 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ #### Features +- Add column-level Dynamic Data Masking (DDM). Declare masks via a first-class `masked_with:` column property in schema YAML or a model-level `masks` ({column: function}) config (which merges key-wise across `dbt_project.yml`, `.yml` and in-file `config()`), and the adapter (re)applies them on every build so they survive full-refresh rebuilds. Diffs against `sys.masked_columns` and emits only changed `ADD`/`MASKED WITH`/`DROP MASKED` DDL; column-level wins over model-level (warned); `masked_with: null` opts a column out of an inherited default; unknown columns are skipped with a warning; unmaskable column types error. Applies to `table`/`incremental`/`snapshot` base tables only (not views or seeds), and masks are applied before index creation so masked index-key columns work on fresh builds. Requires SQL Server 2016+. - Add Postgres-style `indexes` model config for tables, incrementals, seeds and snapshots, covering most `CREATE INDEX` options. [#535](https://github.com/dbt-msft/dbt-sqlserver/issues/535) - Index names are deterministic definition hashes (`dbt_idx_` prefix); creation is idempotent and unchanged definitions are never rebuilt. - Reconcile indexes against the config on incremental, DML-refresh and snapshot runs, applied as one atomic batch. Constraint-backing, legacy post-hook and `as_columnstore` indexes are never dropped. diff --git a/README.md b/README.md index c17e8751..9fa754c2 100644 --- a/README.md +++ b/README.md @@ -216,6 +216,53 @@ You can also set it per model: {{ config(materialized="table", as_columnstore=false) }} ``` +### Dynamic Data Masking (`masked_with` / `masks`) + +The adapter can apply SQL Server [Dynamic Data Masking](https://learn.microsoft.com/en-us/sql/relational-databases/security/dynamic-data-masking) (DDM) to columns as part of the materialization, so masks are re-applied on every build and survive dbt's drop-and-recreate on a full refresh. A principal granted `SELECT` but not `UNMASK` then sees masked values instead of real data (dbt's own build principal, being `db_owner`, keeps `UNMASK` and reads real data). Requires **SQL Server 2016+**. + +There are two config surfaces, and you can use either or both: + +**Column-level `masked_with:`** — a first-class column property in schema YAML (like `data_type:` or `constraints:`), whose value is the masking-function string: + +```yaml +# models/schema.yml +version: 2 +models: + - name: core_patients + columns: + - name: surname + masked_with: "default()" + - name: nhs_number + masked_with: 'partial(0,"XXXXXXXXXX",0)' +``` + +**Model-level `masks`** — a `{column: function}` dict, settable in the in-file `{{ config() }}`, the model's `.yml` `config:` block, or a directory-wide default in `dbt_project.yml`. It merges key-wise across those levels (like `meta`), so a directory default and a per-model tweak combine rather than clobber: + +```sql +{{ config(masks={'surname': "default()", 'nhs_number': 'partial(0,"XXXXXXXXXX",0)'}) }} +``` + +```yaml +# dbt_project.yml — mask nhs_number on every model under datasets/ that has it +models: + my_project: + datasets: + +masks: { nhs_number: "default()" } +``` + +Behaviour: + +- **Precedence:** when both surfaces target the same column, the column-level `masked_with` wins, and a warning naming the model, column and both functions is emitted (even when they agree). This is not something dbt itself ranks, so the rule is the adapter's: a column is more specific than a model. +- **Opt out of an inherited default:** set `masked_with: null` on the column to remove a mask inherited from a directory/model-level `masks` entry. +- **Validation:** a `masks` (or `masked_with`) entry naming a column that is not in the built relation is skipped with a warning (a likely typo or stale rename); the run does not fail. +- **Unmaskable columns:** computed, `FILESTREAM`, sparse `COLUMN_SET` and `Always Encrypted` columns cannot carry a mask, and the run errors listing them rather than emitting DDL that fails. +- **Views/ephemeral/seeds:** masks apply to base tables only (`table`, `incremental`, `snapshot`). Views inherit masking from their base tables and cannot carry a mask; **seeds are not currently masked**. +- **Idempotent:** the adapter diffs the desired masks against `sys.masked_columns` and emits only the `ADD` / change / `DROP MASKED` statements that changed, so a persisted (incremental) re-run with no config change issues no DDL. + +**Indexes and masking.** SQL Server cannot *add* a mask to a column an index depends on (documented for all versions: `ALTER TABLE ALTER COLUMN … failed because one or more objects access this column`). The adapter avoids this on fresh builds by applying masks **before** it creates (rowstore) indexes — which is exactly Microsoft's documented workaround order (mask, then create the index). The default clustered columnstore index is unaffected (its columns are included, not key columns). On a **persisted** table (incremental/snapshot without full refresh), adding a *new* mask to a column that is already an index key errors with a message pointing to the drop-index → mask → recreate-index workaround. + +**Version notes.** All masking DDL the adapter emits (`ADD MASKED`, `MASKED WITH`, `DROP MASKED`) and the functions `default()`, `email()`, `random(a,b)` and `partial(...)` work on 2016+. The `datetime()` partial-date function and granular column/schema/table-scoped `UNMASK` are SQL Server 2022+ only; the adapter never emits them, but mask-function strings are passed through verbatim, so using a 2022-only function on an older server will be rejected by SQL Server. + ## Contributing [![Unit tests](https://github.com/dbt-msft/dbt-sqlserver/actions/workflows/unit-tests.yml/badge.svg)](https://github.com/dbt-msft/dbt-sqlserver/actions/workflows/unit-tests.yml) diff --git a/dbt/adapters/sqlserver/sqlserver_adapter.py b/dbt/adapters/sqlserver/sqlserver_adapter.py index 50629c19..e2ea95dd 100644 --- a/dbt/adapters/sqlserver/sqlserver_adapter.py +++ b/dbt/adapters/sqlserver/sqlserver_adapter.py @@ -1,4 +1,4 @@ -from typing import Any, List, Optional +from typing import Any, Dict, List, Optional import agate import dbt_common.exceptions @@ -29,6 +29,9 @@ from dbt.adapters.sqlserver.sqlserver_column import SQLServerColumn, SQLServerColumnNative from dbt.adapters.sqlserver.sqlserver_configs import SQLServerConfigs from dbt.adapters.sqlserver.sqlserver_connections import SQLServerConnectionManager +from dbt.adapters.sqlserver.sqlserver_mask import ColumnMask +from dbt.adapters.sqlserver.sqlserver_mask import mask_changes as _mask_changes +from dbt.adapters.sqlserver.sqlserver_mask import resolve_masks as _resolve_masks from dbt.adapters.sqlserver.sqlserver_relation import SQLServerRelation logger = AdapterLogger("SQLServer") @@ -502,6 +505,64 @@ def index_changes( "warnings": warnings, } + @available + def resolve_masks(self, model: Any, model_masks: Optional[dict] = None) -> Dict[str, str]: + """Merge the column-level `masked_with` and model-level `masks` surfaces + into one `{column: function}` map for `apply_masks`. + + `model` is the Jinja `model` dict (`node.to_dict()`), whose `columns` + carry any `masked_with` as a flattened key (an explicit `masked_with: + null` survives serialization as a present `None`, signalling opt-out). + `model_masks` is `config.get('masks')` — already surface-merged by dbt. + Precedence and conflict warnings are handled here; key existence is + validated later in the macro against the real relation. + """ + model = model or {} + columns = model.get("columns") or {} + column_masks = [] + for name, col in columns.items(): + col = col or {} + column_masks.append( + ColumnMask( + name=col.get("name", name), + masked_with_present=("masked_with" in col), + masked_with=col.get("masked_with"), + ) + ) + model_name = model.get("name") or model.get("alias") or "" + mask_map, warnings = _resolve_masks(column_masks, model_masks, model_name) + for warning in warnings: + logger.warning(warning) + return mask_map + + @available + def mask_changes( + self, + existing_masks: Any, + mask_config: Optional[dict], + index_key_columns: Any = None, + existing_columns: Any = None, + ) -> dict: + """Diff a resolved mask map against current `sys.masked_columns` state. + + `existing_masks` is the agate table from `get_show_mask_sql` (columns + `name`, `masking_function`). Returns plain lists for jinja: `adds` / + `changes` (each `[column, function]`), `drops` (column names), `skipped` + (warnings for columns absent from the relation) and `errors` (an ADD onto + a current index-key column, which SQL Server rejects). The macro emits + DDL for adds/changes/drops, logs `skipped`, and raises on `errors`.""" + rows = [] + if existing_masks is not None: + column_names = existing_masks.column_names + for row in existing_masks.rows: + rows.append(dict(zip(column_names, row))) + return _mask_changes( + rows, + mask_config or {}, + set(index_key_columns or []), + existing_columns=(list(existing_columns) if existing_columns is not None else None), + ) + COLUMNS_EQUAL_SQL = """ with diff_count as ( diff --git a/dbt/adapters/sqlserver/sqlserver_configs.py b/dbt/adapters/sqlserver/sqlserver_configs.py index f0277f22..ceb8fce8 100644 --- a/dbt/adapters/sqlserver/sqlserver_configs.py +++ b/dbt/adapters/sqlserver/sqlserver_configs.py @@ -1,5 +1,7 @@ -from dataclasses import dataclass -from typing import Any, Optional, Tuple +from dataclasses import dataclass, field +from typing import Any, Dict, Optional, Tuple + +from dbt_common.contracts.config.base import MergeBehavior from dbt.adapters.protocol import AdapterConfig from dbt.adapters.sqlserver.relation_configs import SQLServerIndexConfig @@ -14,3 +16,12 @@ class SQLServerConfigs(AdapterConfig): # false (default) | warn | true - how index reconciliation treats # droppable indexes dbt didn't create (YAML may supply bool or str) drop_unmanaged_indexes: Optional[Any] = False + # column-name -> DDM masking-function map for the model-level `masks` + # surface. MergeBehavior.Update makes it merge key-wise across the config + # chain (dbt_project.yml +masks defaults, .yml config, in-file config()) + # the same way `meta` composes, rather than the default clobber — so a + # directory-level default and a per-model tweak combine instead of one + # replacing the whole dict. + masks: Optional[Dict[str, Any]] = field( + default_factory=dict, metadata=MergeBehavior.Update.meta() + ) diff --git a/dbt/adapters/sqlserver/sqlserver_mask.py b/dbt/adapters/sqlserver/sqlserver_mask.py new file mode 100644 index 00000000..f187c603 --- /dev/null +++ b/dbt/adapters/sqlserver/sqlserver_mask.py @@ -0,0 +1,199 @@ +"""Pure resolution + diff logic for Dynamic Data Masking (DDM). + +Kept free of any database or dbt-context dependency so it can be unit tested +in isolation, mirroring ``relation_configs/index.py``. The adapter's +``@available`` wrappers extract plain data from the model / ``sys.masked_columns`` +and delegate here; the Jinja ``sqlserver__apply_masks`` macro turns the diff +into DDL. + +Two config surfaces feed one per-column mask map: + +* column-level ``masked_with:`` property (a value, or ``null`` to opt out), and +* the model-level ``masks`` dict. + +Column-level wins over model-level, and any conflict is surfaced as a warning. +Column identifiers are compared case-insensitively, matching SQL Server's +default collation. +""" + +from dataclasses import dataclass +from typing import Dict, List, Optional, Sequence, Set, Tuple + + +@dataclass(frozen=True) +class ColumnMask: + """A column's ``masked_with`` declaration, as parsed from schema YAML. + + ``masked_with_present`` distinguishes "the key was written" from "absent": + an explicit ``masked_with: null`` (present with ``masked_with=None``) means + "no mask here", overriding any inherited model-level ``masks`` entry. + """ + + name: str + masked_with_present: bool + masked_with: Optional[str] + + +def _normalize_name(name: str) -> str: + return name.strip().lower() + + +def _pop_ci(mapping: Dict[str, str], name: str) -> Optional[str]: + """Remove a key matching ``name`` case-insensitively; return its value.""" + target = _normalize_name(name) + for key in list(mapping): + if _normalize_name(key) == target: + return mapping.pop(key) + return None + + +def _find_ci(mapping: Dict[str, str], name: str) -> Optional[str]: + """Return the existing key matching ``name`` case-insensitively, if any.""" + target = _normalize_name(name) + for key in mapping: + if _normalize_name(key) == target: + return key + return None + + +def resolve_masks( + column_masks: Sequence[ColumnMask], + model_masks: Optional[Dict[str, str]], + model_name: str, +) -> Tuple[Dict[str, str], List[str]]: + """Merge the two config surfaces into one ``{column: function}`` map. + + ``model_masks`` seeds the map (it is already surface-merged by dbt across + ``dbt_project.yml`` / ``.yml`` / in-file ``config()``). Each column-level + ``masked_with`` then overrides: + + * a function value wins over any model-level entry for the same column and, + when both surfaces target that column, emits a conflict warning (even when + the two functions agree — a duplicate declaration is worth surfacing); + * an explicit ``null`` removes any inherited model-level entry ("opt out"). + + Key *existence* is not validated here — that happens in the macro against + the actually-built relation, which is authoritative and complete even when + the YAML ``columns:`` block is partial. + """ + mask_map: Dict[str, str] = dict(model_masks or {}) + warnings: List[str] = [] + + for cm in column_masks: + if not cm.masked_with_present: + continue + + if cm.masked_with is None: + # explicit opt-out: drop any inherited model-level mask + _pop_ci(mask_map, cm.name) + continue + + conflict_key = _find_ci(model_masks or {}, cm.name) + if conflict_key is not None: + warnings.append( + f"On model '{model_name}', column '{cm.name}' is masked by both a " + f"column-level `masked_with` ('{cm.masked_with}') and a model-level " + f"`masks` entry ('{(model_masks or {})[conflict_key]}'). The column-level " + f"function ('{cm.masked_with}') is applied (column-level wins)." + ) + + # column-level wins: drop any case-variant model-level key, then set + _pop_ci(mask_map, cm.name) + mask_map[cm.name] = cm.masked_with + + return mask_map, warnings + + +def _normalize_function(function: str) -> str: + """Canonicalise a masking-function string for comparison only. + + ``sys.masked_columns.masking_function`` may store a function with different + internal spacing than the user wrote (e.g. ``partial(0, "X", 0)`` vs + ``partial(0,"X",0)``), so whitespace is stripped and case folded before + comparing desired against current. The user's original string is always the + one emitted in DDL. + """ + return "".join(function.split()).lower() + + +def mask_changes( + existing_masks: Sequence[Dict[str, str]], + desired: Dict[str, str], + index_key_columns: Set[str], + existing_columns: Optional[Sequence[str]] = None, +) -> Dict[str, list]: + """Diff the desired mask map against current ``sys.masked_columns`` state. + + ``existing_masks`` is a sequence of ``{"name", "masking_function"}`` rows. + ``existing_columns``, when given, is every column of the built relation; any + desired mask naming a column that is not present is skipped (a typo or a + stale rename) and reported in ``skipped`` rather than emitted as DDL that + would fail. Passing ``None`` disables that check. + + Returns lists keyed ``adds`` / ``changes`` / ``drops`` / ``skipped`` / + ``errors``: + + * ``adds`` – ``[(column, function)]`` masked in config, not yet in the DB + (``ALTER COLUMN ... ADD MASKED``); + * ``changes`` – ``[(column, function)]`` masked in the DB with a different + function (``ALTER COLUMN ... MASKED WITH``, no ``ADD``); + * ``drops`` – ``[column]`` masked in the DB but no longer in config + (``ALTER COLUMN ... DROP MASKED``); + * ``skipped`` – warning strings for desired columns absent from the relation; + * ``errors`` – ``ADD`` targeting a column that is currently an index key, + which SQL Server rejects with a dependency error (documented for all + versions 2016+). Only *adds* hit this: re-specifying the function on an + already-masked index-key column is fine. On fresh builds masks are applied + before indexes exist, so ``index_key_columns`` is empty there and this + never triggers; it only guards the persisted path where an index already + references the column. + """ + existing_by_name = { + _normalize_name(row["name"]): row["masking_function"] for row in existing_masks + } + desired_by_name = {_normalize_name(col): (col, fn) for col, fn in desired.items()} + index_keys = {_normalize_name(c) for c in index_key_columns} + known_columns = ( + {_normalize_name(c) for c in existing_columns} if existing_columns is not None else None + ) + + adds: List[Tuple[str, str]] = [] + changes: List[Tuple[str, str]] = [] + drops: List[str] = [] + skipped: List[str] = [] + errors: List[str] = [] + + for norm, (col, fn) in desired_by_name.items(): + if known_columns is not None and norm not in known_columns: + skipped.append( + f"Column '{col}' is configured for masking but is not a column of " + f"the built relation; skipping. Check for a typo or a renamed column." + ) + continue + if norm not in existing_by_name: + if norm in index_keys: + errors.append( + f"Column '{col}' is configured for masking but is also an index " + f"key column. SQL Server cannot add a mask to a column that an " + f"index depends on. Remove the mask, or remove the column from the " + f"index (drop the index, apply the mask, then recreate the index)." + ) + continue + adds.append((col, fn)) + elif _normalize_function(existing_by_name[norm]) != _normalize_function(fn): + changes.append((col, fn)) + + for norm, current_fn in existing_by_name.items(): + if norm not in desired_by_name: + # preserve the DB's spelling of the column name for the DROP + drops.append( + next(row["name"] for row in existing_masks if _normalize_name(row["name"]) == norm) + ) + + return { + "adds": adds, + "changes": changes, + "drops": drops, + "skipped": skipped, + "errors": errors, + } diff --git a/dbt/include/sqlserver/macros/adapters/apply_masks.sql b/dbt/include/sqlserver/macros/adapters/apply_masks.sql new file mode 100644 index 00000000..2554656e --- /dev/null +++ b/dbt/include/sqlserver/macros/adapters/apply_masks.sql @@ -0,0 +1,170 @@ +{#- + Column-level Dynamic Data Masking (DDM), modelled on apply_grants. + + Post-materialization step: (re)apply the masks a model declares so they + survive dbt's drop-and-recreate on every full refresh. Reads the current + state from sys.masked_columns, diffs against the resolved desired map, and + emits only the ALTERs that changed. A no-op when nothing is configured, on + non-table relations, and on adapters other than SQL Server. + + Config surfaces (merged + precedence-resolved in adapter.resolve_masks): + * column-level `masked_with:` property in schema YAML, and + * model-level `masks` dict config. +-#} + +{% macro apply_masks(relation, mask_config) %} + {{ return(adapter.dispatch('apply_masks', 'dbt')(relation, mask_config)) }} +{% endmacro %} + +{#- Non-SQL-Server adapters (and SQL Server < 2016) are unaffected. -#} +{% macro default__apply_masks(relation, mask_config) %}{% endmacro %} + + +{% macro get_show_mask_sql(relation) %} + {{ return(adapter.dispatch('get_show_mask_sql', 'dbt')(relation)) }} +{% endmacro %} + +{% macro default__get_show_mask_sql(relation) %} + {{ return('') }} +{% endmacro %} + +{% macro sqlserver__get_show_mask_sql(relation) %} + select + c.name as name, + c.masking_function as masking_function + from sys.masked_columns c {{ information_schema_hints() }} + where c.object_id = OBJECT_ID('{{ relation.schema }}.{{ relation.identifier }}') +{% endmacro %} + + +{#- Key columns of any index on the relation. Included columns (is_included + _column = 1) are excluded, which also excludes the default clustered + columnstore index (it reports every column as included, never as a key), so + a normal columnstore table has no index-key columns and masks apply freely. -#} +{% macro sqlserver__get_mask_index_key_columns(relation) %} + {% call statement('get_mask_index_key_columns', fetch_result=True) %} + select distinct col.name as name + from sys.index_columns ic {{ information_schema_hints() }} + inner join sys.columns col {{ information_schema_hints() }} + on col.object_id = ic.object_id and col.column_id = ic.column_id + where ic.object_id = OBJECT_ID('{{ relation.schema }}.{{ relation.identifier }}') + and ic.is_included_column = 0 + {% endcall %} + {% set result = [] %} + {% for row in load_result('get_mask_index_key_columns').table.rows %} + {% do result.append(row[0]) %} + {% endfor %} + {{ return(result) }} +{% endmacro %} + + +{#- Columns that DDM cannot mask at all (a mask ALTER would fail): computed, + FILESTREAM, sparse COLUMN_SET, and Always Encrypted columns. -#} +{% macro sqlserver__get_unmaskable_columns(relation) %} + {% call statement('get_unmaskable_columns', fetch_result=True) %} + select col.name as name + from sys.columns col {{ information_schema_hints() }} + where col.object_id = OBJECT_ID('{{ relation.schema }}.{{ relation.identifier }}') + and ( + col.is_computed = 1 + or col.is_filestream = 1 + or col.is_column_set = 1 + or col.encryption_type is not null + ) + {% endcall %} + {% set result = [] %} + {% for row in load_result('get_unmaskable_columns').table.rows %} + {% do result.append(row[0]) %} + {% endfor %} + {{ return(result) }} +{% endmacro %} + + +{% macro sqlserver__apply_masks(relation, mask_config) %} + {#-- If mask_config is {} or None, this is a no-op (mirrors apply_grants). --#} + {% if not mask_config %} + {{ return(none) }} + {% endif %} + + {#-- DDM applies to base tables only. Views/ephemeral inherit masking from + their base tables but cannot carry an ALTER-ed mask, so no-op. --#} + {% if relation.type != 'table' %} + {{ return(none) }} + {% endif %} + + {#-- DDM requires SQL Server 2016 (major 13)+ ("Applies to: SQL Server 2016 + (13.x) and later"); sys.masked_columns does not exist before then. Fail + clearly rather than with a cryptic error. --#} + {% set major = sqlserver__server_major_version() %} + {% if major is not none and major < 13 %} + {{ exceptions.raise_compiler_error( + "Dynamic Data Masking (the `masks` / `masked_with` config) requires SQL " + ~ "Server 2016 (major version 13) or newer; detected major version " + ~ major ~ " for " ~ relation ~ ". Remove the mask configuration.") }} + {% endif %} + + {#-- Current mask state and the metadata needed to validate the desired map. --#} + {% set existing_masks = run_query(get_show_mask_sql(relation)) %} + {% set existing_columns = adapter.get_columns_in_relation(relation) + | map(attribute='name') | list %} + {% set index_key_columns = sqlserver__get_mask_index_key_columns(relation) %} + {% set unmaskable_columns = sqlserver__get_unmaskable_columns(relation) %} + + {% set changes = adapter.mask_changes( + existing_masks, mask_config, index_key_columns, existing_columns) %} + + {#-- Surface skipped columns (typo / renamed) as warnings, don't fail. --#} + {% for message in changes['skipped'] %} + {% do exceptions.warn("apply_masks on " ~ relation ~ ": " ~ message) %} + {% endfor %} + + {#-- Unmaskable column types: raise rather than let the ALTER fail. --#} + {% set unmaskable_lower = unmaskable_columns | map('lower') | list %} + {% set unmaskable_hits = [] %} + {% for col, fn in changes['adds'] + changes['changes'] %} + {% if col | lower in unmaskable_lower %} + {% do unmaskable_hits.append(col) %} + {% endif %} + {% endfor %} + {% if unmaskable_hits %} + {{ exceptions.raise_compiler_error( + "apply_masks on " ~ relation ~ ": cannot mask column(s) " + ~ (unmaskable_hits | join(", ")) ~ " — computed, FILESTREAM, sparse " + ~ "COLUMN_SET and Always Encrypted columns cannot carry a mask.") }} + {% endif %} + + {#-- Index-key collision (SQL Server < 2022): raise with a descriptive + message instead of emitting DDL that fails mid-transaction. --#} + {% if changes['errors'] %} + {{ exceptions.raise_compiler_error( + "apply_masks on " ~ relation ~ ":\n" ~ (changes['errors'] | join("\n"))) }} + {% endif %} + + {#-- Emit only what changed. --#} + {% set statements = [] %} + {% for col, fn in changes['adds'] %} + {% do statements.append( + "alter table " ~ relation ~ " alter column [" + ~ (col | replace(']', ']]')) ~ "] add masked with (function = '" + ~ (fn | replace("'", "''")) ~ "')") %} + {% endfor %} + {% for col, fn in changes['changes'] %} + {% do statements.append( + "alter table " ~ relation ~ " alter column [" + ~ (col | replace(']', ']]')) ~ "] masked with (function = '" + ~ (fn | replace("'", "''")) ~ "')") %} + {% endfor %} + {% for col in changes['drops'] %} + {% do statements.append( + "alter table " ~ relation ~ " alter column [" + ~ (col | replace(']', ']]')) ~ "] drop masked") %} + {% endfor %} + + {% if statements %} + {% do run_query(statements | join(";\n")) %} + {% do log("Applied " ~ statements | length ~ " data-mask change(s) on " + ~ relation, info=true) %} + {% else %} + {% do log("On " ~ relation ~ ": all data masks are in place, no changes needed.") %} + {% endif %} +{% endmacro %} diff --git a/dbt/include/sqlserver/macros/materializations/models/incremental/incremental.sql b/dbt/include/sqlserver/macros/materializations/models/incremental/incremental.sql index 8c60ba25..1bfbbe5d 100644 --- a/dbt/include/sqlserver/macros/materializations/models/incremental/incremental.sql +++ b/dbt/include/sqlserver/macros/materializations/models/incremental/incremental.sql @@ -79,11 +79,17 @@ {% do persist_docs(target_relation, model) %} + {% set mask_config = adapter.resolve_masks(model, config.get('masks')) %} {% if existing_relation is none or existing_relation.is_view or should_full_refresh() %} + {# Freshly built table: mask before creating (rowstore) indexes, since a + mask cannot be added to a column an index depends on (all versions). #} + {% do apply_masks(target_relation, mask_config) %} {% do create_indexes(target_relation) %} {% else %} - {# Table persisted across this run: converge its indexes on the config. #} + {# Table persisted across this run: converge its indexes on the config, + then reconcile masks (index drops land first). #} {% do sqlserver__reconcile_indexes(target_relation) %} + {% do apply_masks(target_relation, mask_config) %} {% endif %} {{ run_hooks(post_hooks, inside_transaction=True) }} diff --git a/dbt/include/sqlserver/macros/materializations/models/table/table.sql b/dbt/include/sqlserver/macros/materializations/models/table/table.sql index 9d3375d7..367b94af 100644 --- a/dbt/include/sqlserver/macros/materializations/models/table/table.sql +++ b/dbt/include/sqlserver/macros/materializations/models/table/table.sql @@ -59,6 +59,15 @@ {{ adapter.rename_relation(intermediate_relation, target_relation) }} + {#-- Apply data masks before create_indexes: a mask cannot be added to a + column an index depends on (documented for all SQL Server versions; + the fix is to mask first, then create the index — exactly this order), + so masking must happen while the (rowstore) indexes do not yet exist. + The clustered columnstore index built during CTAS is fine — columnstore + columns are reported as included, not index keys, and can be masked. --#} + {% set mask_config = adapter.resolve_masks(model, config.get('masks')) %} + {% do apply_masks(target_relation, mask_config) %} + {% do create_indexes(target_relation) %} {% endif %} diff --git a/dbt/include/sqlserver/macros/materializations/models/table/table_dml_refresh.sql b/dbt/include/sqlserver/macros/materializations/models/table/table_dml_refresh.sql index 78cd4687..38343dde 100644 --- a/dbt/include/sqlserver/macros/materializations/models/table/table_dml_refresh.sql +++ b/dbt/include/sqlserver/macros/materializations/models/table/table_dml_refresh.sql @@ -46,6 +46,10 @@ {%- set schema_changes = check_for_schema_changes(refresh_relation, target_relation) -%} {%- set schema_match = not schema_changes['schema_changed'] -%} + {#- Resolve the mask map once; applied per-branch below at the right point + relative to index creation (see apply_masks). -#} + {%- set mask_config = adapter.resolve_masks(model, config.get('masks')) -%} + {% if schema_match %} {# Use the target's physical column order for both INSERT and SELECT. #} {# The scratch table has the same columns but possibly in a different order, #} @@ -78,6 +82,11 @@ the config. Runs after the swap's self-contained transaction. #} {% do sqlserver__reconcile_indexes(target_relation) %} + {# Persisted-table path: masks already exist from the prior build; this + reconciles any config change. Runs after reconcile so index drops land + first. #} + {% do apply_masks(target_relation, mask_config) %} + {% else %} {# Schema changed — fall back to rename-swap for this run #} {{ log("Schema change detected for " ~ target_relation ~ " — falling back to rename-swap", info=true) }} @@ -94,6 +103,11 @@ {{ adapter.rename_relation(refresh_relation, target_relation) }} + {# Rebuilt via SELECT INTO (no masks carried), so apply masks before + create_indexes — a mask cannot be added to a column an index depends + on (documented for all SQL Server versions). #} + {% do apply_masks(target_relation, mask_config) %} + {% do create_indexes(target_relation) %} {{ drop_relation_if_exists(backup_relation) }} diff --git a/dbt/include/sqlserver/macros/materializations/snapshots/snapshot.sql b/dbt/include/sqlserver/macros/materializations/snapshots/snapshot.sql index ac601481..9aa6095d 100644 --- a/dbt/include/sqlserver/macros/materializations/snapshots/snapshot.sql +++ b/dbt/include/sqlserver/macros/materializations/snapshots/snapshot.sql @@ -109,11 +109,17 @@ {% do persist_docs(target_relation, model) %} + {% set mask_config = adapter.resolve_masks(model, config.get('masks')) %} {% if not target_relation_exists %} + {# Freshly built snapshot table: mask before creating (rowstore) indexes, + since a mask cannot be added to a column an index depends on (all versions). #} + {% do apply_masks(target_relation, mask_config) %} {% do create_indexes(target_relation) %} {% else %} - {# Snapshot table persisted: converge its indexes on the config. #} + {# Snapshot table persisted: converge its indexes on the config, then + reconcile masks (index drops land first). #} {% do sqlserver__reconcile_indexes(target_relation) %} + {% do apply_masks(target_relation, mask_config) %} {% endif %} {{ run_hooks(post_hooks, inside_transaction=True) }} diff --git a/tests/functional/adapter/mssql/test_masks.py b/tests/functional/adapter/mssql/test_masks.py new file mode 100644 index 00000000..232ccd1b --- /dev/null +++ b/tests/functional/adapter/mssql/test_masks.py @@ -0,0 +1,335 @@ +"""Functional tests for column-level Dynamic Data Masking (DDM). + +Exercises the two config surfaces (column-level ``masked_with:`` and the +model-level ``masks`` dict), precedence/opt-out/validation, the introspection +against ``sys.masked_columns``, that masks survive a full-refresh rebuild, and +that an un-privileged (no ``UNMASK``) principal actually sees masked values. + +Requires SQL Server 2016+ (DDM). The CI/test server is 2022. +""" + +import pytest + +from dbt.tests.util import get_connection, run_dbt, run_dbt_and_capture + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + + +def masked_columns(project, table_name): + """Return {column_name: masking_function} from sys.masked_columns.""" + sql = f""" + select c.name, c.masking_function + from sys.masked_columns c + where c.object_id = OBJECT_ID('{project.test_schema}.{table_name}') + """ + with get_connection(project.adapter): + _, table = project.adapter.execute(sql, fetch=True) + return {row[0]: row[1] for row in table.rows} + + +def select_as_unprivileged(project, table_name, columns): + """Read `columns` from the table as a freshly-created user that holds + SELECT but not UNMASK, so masking is in effect. Returns the first row.""" + cols = ", ".join(columns) + user = "dbt_mask_reader" + sql = f""" + if database_principal_id('{user}') is null + create user {user} without login; + grant select on schema::{project.test_schema} to {user}; + execute as user = '{user}'; + select {cols} from {project.test_schema}.{table_name}; + revert; + """ + with get_connection(project.adapter): + _, table = project.adapter.execute(sql, fetch=True) + return table.rows[0] + + +# --------------------------------------------------------------------------- +# model fixtures +# --------------------------------------------------------------------------- + +# column-level `masked_with:` surface +column_property_model_sql = """ +{{ config(materialized="table") }} +select + 1 as id, + cast('Smith' as varchar(50)) as surname, + cast('1234567890' as varchar(10)) as nhs_number +""" + +column_property_yml = """ +version: 2 +models: + - name: masked_model + columns: + - name: surname + masked_with: "default()" + - name: nhs_number + masked_with: 'partial(0,"XXXXXXXXXX",0)' +""" + +# model-level `masks` dict surface +model_level_masks_sql = """ +{{ config( + materialized="table", + masks={"surname": "default()", "nhs_number": "email()"} +) }} +select + 1 as id, + cast('Smith' as varchar(50)) as surname, + cast('a@b.com' as varchar(50)) as nhs_number +""" + + +# --------------------------------------------------------------------------- +# column-level `masked_with:` surface +# --------------------------------------------------------------------------- + + +class TestColumnPropertyMasks: + @pytest.fixture(scope="class") + def models(self): + return { + "masked_model.sql": column_property_model_sql, + "masked_model.yml": column_property_yml, + } + + def test_masks_applied_and_survive_full_refresh(self, project): + run_dbt(["run"]) + masks = masked_columns(project, "masked_model") + assert masks.get("surname") == "default()" + assert masks.get("nhs_number") == 'partial(0, "XXXXXXXXXX", 0)' + assert "id" not in masks + + # a full refresh drops & recreates the table — the whole point is that + # the adapter re-applies the mask so it is not lost + run_dbt(["run", "--full-refresh"]) + masks = masked_columns(project, "masked_model") + assert masks.get("surname") == "default()" + assert masks.get("nhs_number") == 'partial(0, "XXXXXXXXXX", 0)' + + def test_unprivileged_user_sees_masked_values(self, project): + run_dbt(["run"]) + surname, nhs = select_as_unprivileged(project, "masked_model", ["surname", "nhs_number"]) + # default() masks a string to 'xxxx'; partial(0,"XXXXXXXXXX",0) keeps + # only the padding + assert surname == "xxxx" + assert nhs == "XXXXXXXXXX" + + +# --------------------------------------------------------------------------- +# model-level `masks` surface +# --------------------------------------------------------------------------- + + +class TestModelLevelMasks: + @pytest.fixture(scope="class") + def models(self): + return {"masked_model.sql": model_level_masks_sql} + + def test_model_level_masks_applied(self, project): + run_dbt(["run"]) + masks = masked_columns(project, "masked_model") + assert masks.get("surname") == "default()" + assert masks.get("nhs_number") == "email()" + + +# --------------------------------------------------------------------------- +# snapshot materialization +# --------------------------------------------------------------------------- + +snapshot_source_model_sql = """ +{{ config(materialized="table") }} +select 1 as id, cast('Smith' as varchar(50)) as surname +""" + +masked_snapshot_sql = """ +{% snapshot masked_snapshot %} +{{ config( + unique_key='id', + strategy='check', + check_cols=['surname'], + masks={'surname': 'default()'} +) }} +select * from {{ ref('snap_source') }} +{% endsnapshot %} +""" + + +class TestSnapshotMasks: + @pytest.fixture(scope="class") + def models(self): + return {"snap_source.sql": snapshot_source_model_sql} + + @pytest.fixture(scope="class") + def snapshots(self): + return {"masked_snapshot.sql": masked_snapshot_sql} + + def test_masks_applied_to_snapshot_table(self, project): + run_dbt(["run"]) + run_dbt(["snapshot"]) + masks = masked_columns(project, "masked_snapshot") + assert masks.get("surname") == "default()" + # dbt-maintained snapshot metadata columns are left unmasked + assert "id" not in masks + + # a second snapshot run persists the table — mask stays, no error + run_dbt(["snapshot"]) + assert masked_columns(project, "masked_snapshot").get("surname") == "default()" + + +# --------------------------------------------------------------------------- +# idempotency / change / drop +# --------------------------------------------------------------------------- + + +incremental_masked_sql = """ +{{ config(materialized="incremental", masks={"surname": "default()"}) }} +select 1 as id, cast('Smith' as varchar(50)) as surname +{% if is_incremental() %} where 1 = 0 {% endif %} +""" + + +class TestMaskIdempotentOnPersistedTable: + """On a persisted (incremental, non-full-refresh) table the mask survives + the run, so re-applying is a true no-op that emits zero mask DDL.""" + + @pytest.fixture(scope="class") + def models(self): + return {"masked_model.sql": incremental_masked_sql} + + def test_rerun_emits_no_mask_ddl(self, project): + _, first = run_dbt_and_capture(["run"]) + assert "data-mask change" in first # applied on first build + assert masked_columns(project, "masked_model").get("surname") == "default()" + + # second (incremental, persisted) run: mask already in place → no DDL + _, second = run_dbt_and_capture(["run"]) + assert "data-mask change" not in second + assert masked_columns(project, "masked_model").get("surname") == "default()" + + +class TestMaskLifecycle: + @pytest.fixture(scope="class") + def models(self): + return { + "masked_model.sql": column_property_model_sql, + "masked_model.yml": column_property_yml, + } + + def test_change_function_updates(self, project): + run_dbt(["run"]) + # swap surname's function + changed_yml = column_property_yml.replace( + 'masked_with: "default()"', 'masked_with: "email()"' + ) + write_model(project, "masked_model.yml", changed_yml) + run_dbt(["run"]) + assert masked_columns(project, "masked_model").get("surname") == "email()" + + def test_removing_config_drops_mask(self, project): + run_dbt(["run"]) + assert "surname" in masked_columns(project, "masked_model") + # drop the yml entirely — no masks configured anymore + write_model(project, "masked_model.yml", "version: 2\nmodels: []\n") + run_dbt(["run"]) + assert masked_columns(project, "masked_model") == {} + + +# --------------------------------------------------------------------------- +# precedence, opt-out, validation +# --------------------------------------------------------------------------- + +both_surfaces_sql = """ +{{ config(materialized="table", masks={"surname": "default()"}) }} +select 1 as id, cast('Smith' as varchar(50)) as surname +""" + +both_surfaces_yml = """ +version: 2 +models: + - name: masked_model + columns: + - name: surname + masked_with: "email()" +""" + + +class TestBothSurfacesColumnWins: + @pytest.fixture(scope="class") + def models(self): + return { + "masked_model.sql": both_surfaces_sql, + "masked_model.yml": both_surfaces_yml, + } + + def test_column_level_wins_and_warns(self, project): + _, logs = run_dbt_and_capture(["run"]) + # column-level function applied, not the model-level one + assert masked_columns(project, "masked_model").get("surname") == "email()" + # a conflict warning was surfaced naming the model and both functions + assert "masked_model" in logs + assert "email()" in logs and "default()" in logs + assert "column-level" in logs + + +opt_out_sql = """ +{{ config(materialized="table", masks={"surname": "default()", "nhs_number": "default()"}) }} +select 1 as id, cast('Smith' as varchar(50)) as surname, + cast('x' as varchar(50)) as nhs_number +""" + +opt_out_yml = """ +version: 2 +models: + - name: masked_model + columns: + - name: nhs_number + masked_with: null +""" + + +class TestColumnLevelOptOut: + @pytest.fixture(scope="class") + def models(self): + return { + "masked_model.sql": opt_out_sql, + "masked_model.yml": opt_out_yml, + } + + def test_null_opts_out_of_inherited_mask(self, project): + run_dbt(["run"]) + masks = masked_columns(project, "masked_model") + # surname keeps the model-level default; nhs_number opted out + assert masks.get("surname") == "default()" + assert "nhs_number" not in masks + + +invalid_key_sql = """ +{{ config(materialized="table", masks={"surnam": "default()"}) }} +select 1 as id, cast('Smith' as varchar(50)) as surname +""" + + +class TestInvalidModelMaskKey: + @pytest.fixture(scope="class") + def models(self): + return {"masked_model.sql": invalid_key_sql} + + def test_unknown_column_warns_and_is_skipped(self, project): + _, logs = run_dbt_and_capture(["run"]) + # run succeeds, no mask applied, and a warning names the bad column + assert masked_columns(project, "masked_model") == {} + assert "surnam" in logs + + +# helper defined at module end so it is importable above via closure at call time +def write_model(project, filename, contents): + import os + + path = os.path.join(project.project_root, "models", filename) + with open(path, "w") as f: + f.write(contents) diff --git a/tests/unit/adapters/mssql/test_mask.py b/tests/unit/adapters/mssql/test_mask.py new file mode 100644 index 00000000..f60da8e4 --- /dev/null +++ b/tests/unit/adapters/mssql/test_mask.py @@ -0,0 +1,235 @@ +"""Unit tests for the pure mask resolution + diff logic. + +These exercise the surface-merge / precedence / opt-out rules and the +desired-vs-current diff without needing a database connection, mirroring +tests/unit/adapters/mssql/test_index_diff.py. +""" + +from dbt.adapters.sqlserver.sqlserver_mask import ( + ColumnMask, + mask_changes, + resolve_masks, +) + +# --------------------------------------------------------------------------- +# resolve_masks: merge the two config surfaces into one per-column map +# --------------------------------------------------------------------------- + + +def col(name, value, present=True): + return ColumnMask(name=name, masked_with_present=present, masked_with=value) + + +def test_resolve_model_level_only(): + mask_map, warnings = resolve_masks( + column_masks=[], + model_masks={"surname": "default()", "nhs_number": 'partial(0,"X",0)'}, + model_name="core_patients", + ) + assert mask_map == {"surname": "default()", "nhs_number": 'partial(0,"X",0)'} + assert warnings == [] + + +def test_resolve_column_level_only(): + mask_map, warnings = resolve_masks( + column_masks=[col("surname", "default()")], + model_masks={}, + model_name="core_patients", + ) + assert mask_map == {"surname": "default()"} + assert warnings == [] + + +def test_resolve_column_wins_over_model_and_warns(): + mask_map, warnings = resolve_masks( + column_masks=[col("surname", "email()")], + model_masks={"surname": "default()"}, + model_name="core_patients", + ) + assert mask_map == {"surname": "email()"} + assert len(warnings) == 1 + # warning names the model, column, both functions + w = warnings[0] + assert "core_patients" in w + assert "surname" in w + assert "email()" in w and "default()" in w + + +def test_resolve_conflict_warns_even_when_functions_agree(): + _, warnings = resolve_masks( + column_masks=[col("surname", "default()")], + model_masks={"surname": "default()"}, + model_name="core_patients", + ) + assert len(warnings) == 1 + + +def test_resolve_null_opts_out_of_inherited_model_default(): + mask_map, warnings = resolve_masks( + column_masks=[col("nhs_number", None)], + model_masks={"nhs_number": "default()", "surname": "default()"}, + model_name="core_patients", + ) + assert mask_map == {"surname": "default()"} + assert warnings == [] + + +def test_resolve_null_opt_out_with_no_inherited_mask_is_noop(): + mask_map, warnings = resolve_masks( + column_masks=[col("surname", None)], + model_masks={}, + model_name="core_patients", + ) + assert mask_map == {} + assert warnings == [] + + +def test_resolve_is_case_insensitive_for_precedence_and_opt_out(): + # column-level wins even when the model-level key differs only in case + mask_map, warnings = resolve_masks( + column_masks=[col("NHS_Number", "email()")], + model_masks={"nhs_number": "default()"}, + model_name="core_patients", + ) + assert list(mask_map.values()) == ["email()"] + assert len(mask_map) == 1 + assert len(warnings) == 1 + + +# --------------------------------------------------------------------------- +# mask_changes: diff desired map against sys.masked_columns state +# --------------------------------------------------------------------------- + + +def existing(name, fn): + return {"name": name, "masking_function": fn} + + +def test_diff_add_new_mask(): + result = mask_changes( + existing_masks=[], + desired={"surname": "default()"}, + index_key_columns=set(), + ) + assert result["adds"] == [("surname", "default()")] + assert result["changes"] == [] + assert result["drops"] == [] + assert result["errors"] == [] + + +def test_diff_noop_when_unchanged(): + result = mask_changes( + existing_masks=[existing("surname", "default()")], + desired={"surname": "default()"}, + index_key_columns=set(), + ) + assert result["adds"] == [] + assert result["changes"] == [] + assert result["drops"] == [] + + +def test_diff_change_function(): + result = mask_changes( + existing_masks=[existing("surname", "default()")], + desired={"surname": "email()"}, + index_key_columns=set(), + ) + assert result["changes"] == [("surname", "email()")] + assert result["adds"] == [] + assert result["drops"] == [] + + +def test_diff_drop_when_removed_from_config(): + result = mask_changes( + existing_masks=[existing("surname", "default()")], + desired={}, + index_key_columns=set(), + ) + assert result["drops"] == ["surname"] + assert result["adds"] == [] + assert result["changes"] == [] + + +def test_diff_is_case_insensitive_on_column_names(): + result = mask_changes( + existing_masks=[existing("Surname", "default()")], + desired={"surname": "default()"}, + index_key_columns=set(), + ) + assert result["adds"] == [] + assert result["changes"] == [] + assert result["drops"] == [] + + +def test_diff_function_comparison_ignores_whitespace(): + # sys.masked_columns may store partial() with reformatted spacing + result = mask_changes( + existing_masks=[existing("nhs_number", 'partial(0, "XXXXXXXXXX", 0)')], + desired={"nhs_number": 'partial(0,"XXXXXXXXXX",0)'}, + index_key_columns=set(), + ) + assert result["changes"] == [] + assert result["adds"] == [] + + +def test_diff_add_to_index_key_column_is_an_error_not_a_ddl(): + result = mask_changes( + existing_masks=[], + desired={"nhs_number": "default()"}, + index_key_columns={"nhs_number"}, + ) + assert result["adds"] == [] + assert len(result["errors"]) == 1 + assert "nhs_number" in result["errors"][0] + + +def test_diff_changing_existing_mask_on_index_key_is_allowed(): + # the column is already masked, so no ADD dependency problem — a plain + # MASKED WITH change does not hit the index dependency error + result = mask_changes( + existing_masks=[existing("nhs_number", "default()")], + desired={"nhs_number": "email()"}, + index_key_columns={"nhs_number"}, + ) + assert result["changes"] == [("nhs_number", "email()")] + assert result["errors"] == [] + + +# --------------------------------------------------------------------------- +# mask_changes: existence validation against the built relation's columns +# --------------------------------------------------------------------------- + + +def test_diff_skips_and_warns_on_column_absent_from_relation(): + result = mask_changes( + existing_masks=[], + desired={"surnam": "default()"}, # typo — no such column + index_key_columns=set(), + existing_columns=["id", "surname"], + ) + assert result["adds"] == [] + assert result["changes"] == [] + assert len(result["skipped"]) == 1 + assert "surnam" in result["skipped"][0] + + +def test_diff_existence_check_is_case_insensitive(): + result = mask_changes( + existing_masks=[], + desired={"Surname": "default()"}, + index_key_columns=set(), + existing_columns=["id", "surname"], + ) + assert result["adds"] == [("Surname", "default()")] + assert result["skipped"] == [] + + +def test_diff_without_existing_columns_skips_no_validation(): + # existing_columns=None (default) preserves the un-validated diff behaviour + result = mask_changes( + existing_masks=[], + desired={"anything": "default()"}, + index_key_columns=set(), + ) + assert result["adds"] == [("anything", "default()")] + assert result["skipped"] == []