unitycatalog_server/api/
policies.rs

1use itertools::Itertools;
2
3use unitycatalog_common::models::policies::v1::policy_info::Options;
4use unitycatalog_common::models::policies::v1::*;
5use unitycatalog_common::models::{
6    ObjectLabel, Resource, ResourceIdent, ResourceName, ResourceRef,
7};
8
9use super::{RequestContext, SecuredAction, UpdateMask};
10use crate::Result;
11pub use crate::codegen::policies::PolicyHandler;
12use crate::policy::{Permission, Policy, process_resources};
13use crate::store::ResourceStore;
14
15/// The securable types a policy can be defined on, in their canonical (upper-case) form.
16const SECURABLE_TYPES: [&str; 3] = ["CATALOG", "SCHEMA", "TABLE"];
17
18/// Normalize a securable type from a path or body to its canonical upper-case form.
19///
20/// Matching is case-insensitive so `table` and `TABLE` address the same securable.
21fn securable_type(raw: &str) -> Result<&'static str> {
22    SECURABLE_TYPES
23        .into_iter()
24        .find(|t| t.eq_ignore_ascii_case(raw))
25        .ok_or_else(|| {
26            crate::Error::invalid_argument(format!(
27                "unsupported securable type '{raw}': expected CATALOG, SCHEMA or TABLE"
28            ))
29        })
30}
31
32/// A securable a policy can be defined on, together with its ancestors.
33///
34/// Ordered from the securable itself outward: a table's chain is
35/// `[(TABLE, catalog.schema.table), (SCHEMA, catalog.schema), (CATALOG, catalog)]`.
36fn securable_chain(
37    on_securable_type: &str,
38    on_securable_fullname: &str,
39) -> Vec<(&'static str, String)> {
40    let segments: Vec<&str> = on_securable_fullname.split('.').collect();
41    let levels: &[&str] = match on_securable_type {
42        "TABLE" => &["TABLE", "SCHEMA", "CATALOG"],
43        "SCHEMA" => &["SCHEMA", "CATALOG"],
44        "CATALOG" => &["CATALOG"],
45        _ => &[],
46    };
47    levels
48        .iter()
49        .enumerate()
50        .filter_map(|(i, level)| {
51            let take = segments.len().checked_sub(i)?;
52            (take > 0).then(|| (*level, segments[..take].join(".")))
53        })
54        .collect()
55}
56
57fn matches_securable(
58    policy: &PolicyInfo,
59    on_securable_type: &str,
60    on_securable_fullname: &str,
61) -> bool {
62    policy
63        .on_securable_type
64        .eq_ignore_ascii_case(on_securable_type)
65        && policy.on_securable_fullname == on_securable_fullname
66}
67
68fn validate_policy_shape(policy: &PolicyInfo) -> Result<()> {
69    let policy_type = policy
70        .policy_type
71        .as_known()
72        .ok_or_else(|| crate::Error::invalid_argument("unrecognized policy_type"))?;
73    let invalid = |msg: &str| Err(crate::Error::invalid_argument(msg));
74    match (policy_type, &policy.options) {
75        (PolicyType::POLICY_TYPE_UNSPECIFIED, _) => invalid(
76            "policy_type must be one of POLICY_TYPE_ROW_FILTER, POLICY_TYPE_COLUMN_MASK, \
77             POLICY_TYPE_DENY or POLICY_TYPE_GRANT",
78        ),
79        (PolicyType::POLICY_TYPE_ROW_FILTER, Some(Options::RowFilter(r)))
80            if !r.function_name.is_empty() =>
81        {
82            Ok(())
83        }
84        (PolicyType::POLICY_TYPE_ROW_FILTER, _) => {
85            invalid("row filter policies require row_filter.function_name")
86        }
87        (PolicyType::POLICY_TYPE_COLUMN_MASK, _) if policy.match_columns.is_empty() => {
88            invalid("column mask policies require at least one match_columns entry")
89        }
90        (PolicyType::POLICY_TYPE_COLUMN_MASK, Some(Options::ColumnMask(m)))
91            if !m.function_name.is_empty() =>
92        {
93            Ok(())
94        }
95        (PolicyType::POLICY_TYPE_COLUMN_MASK, _) => {
96            invalid("column mask policies require column_mask.function_name")
97        }
98        (PolicyType::POLICY_TYPE_DENY, Some(Options::Deny(_)))
99        | (PolicyType::POLICY_TYPE_GRANT, Some(Options::Grant(_))) => Ok(()),
100        (PolicyType::POLICY_TYPE_DENY, _) => invalid("deny policies require deny options"),
101        (PolicyType::POLICY_TYPE_GRANT, _) => invalid("grant policies require grant options"),
102    }
103}
104
105/// Collapse the per-kind option fields of a create/update request into the stored oneof.
106///
107/// Requests carry `row_filter`, `column_mask`, `deny` and `grant` as separate fields (the
108/// same JSON as the oneof on [`PolicyInfo`]); at most one may be set.
109fn request_options(
110    row_filter: Option<FunctionRef>,
111    column_mask: Option<FunctionRef>,
112    deny: Option<PrivilegeOptions>,
113    grant: Option<PrivilegeOptions>,
114) -> Result<Option<Options>> {
115    let options: Vec<Options> = [
116        row_filter.map(|f| Options::RowFilter(Box::new(f))),
117        column_mask.map(|f| Options::ColumnMask(Box::new(f))),
118        deny.map(|p| Options::Deny(Box::new(p))),
119        grant.map(|p| Options::Grant(Box::new(p))),
120    ]
121    .into_iter()
122    .flatten()
123    .collect();
124    match <[Options; 1]>::try_from(options) {
125        Ok([options]) => Ok(Some(options)),
126        Err(options) if options.is_empty() => Ok(None),
127        Err(_) => Err(crate::Error::invalid_argument(
128            "at most one of row_filter, column_mask, deny and grant may be set",
129        )),
130    }
131}
132
133/// Apply an update request onto the stored policy.
134///
135/// With an empty or `*` update mask every field set in the request is applied. With an
136/// explicit mask only the listed fields are applied, and a listed field that is unset in
137/// the request is cleared. The enforcement options (`row_filter`, `column_mask`, `deny`,
138/// `grant`) are a single oneof, so naming any of them replaces the options as a whole.
139fn apply_update(policy: &mut PolicyInfo, request: UpdatePolicyRequest) -> Result<()> {
140    let options = request_options(
141        request.row_filter.into_option(),
142        request.column_mask.into_option(),
143        request.deny.into_option(),
144        request.grant.into_option(),
145    )?;
146    let mask = UpdateMask::parse(
147        request.update_mask.as_deref(),
148        &[
149            "comment",
150            "to_principals",
151            "except_principals",
152            "for_securable_type",
153            "when_condition",
154            "match_columns",
155            "policy_type",
156            "row_filter",
157            "column_mask",
158            "deny",
159            "grant",
160        ],
161    )?;
162    let masked = |field: &str| mask.names(field);
163
164    if matches!(mask, UpdateMask::All) {
165        if request.comment.is_some() {
166            policy.comment = request.comment;
167        }
168        if !request.to_principals.is_empty() {
169            policy.to_principals = request.to_principals;
170        }
171        if !request.except_principals.is_empty() {
172            policy.except_principals = request.except_principals;
173        }
174        if request.for_securable_type.is_some() {
175            policy.for_securable_type = request.for_securable_type;
176        }
177        if request.when_condition.is_some() {
178            policy.when_condition = request.when_condition;
179        }
180        if !request.match_columns.is_empty() {
181            policy.match_columns = request.match_columns;
182        }
183        if let Some(policy_type) = request.policy_type {
184            policy.policy_type = policy_type;
185        }
186        if options.is_some() {
187            policy.options = options;
188        }
189        return Ok(());
190    }
191
192    if masked("comment") {
193        policy.comment = request.comment;
194    }
195    if masked("to_principals") {
196        policy.to_principals = request.to_principals;
197    }
198    if masked("except_principals") {
199        policy.except_principals = request.except_principals;
200    }
201    if masked("for_securable_type") {
202        policy.for_securable_type = request.for_securable_type;
203    }
204    if masked("when_condition") {
205        policy.when_condition = request.when_condition;
206    }
207    if masked("match_columns") {
208        policy.match_columns = request.match_columns;
209    }
210    if masked("policy_type") {
211        policy.policy_type = request.policy_type.unwrap_or_default();
212    }
213    if ["row_filter", "column_mask", "deny", "grant"]
214        .iter()
215        .any(|f| masked(f))
216    {
217        policy.options = options;
218    }
219    Ok(())
220}
221
222#[async_trait::async_trait]
223impl<T: ResourceStore + Policy<RequestContext>> PolicyHandler<RequestContext> for T {
224    #[tracing::instrument(skip(self, context), fields(resource_name))]
225    async fn create_policy(
226        &self,
227        request: CreatePolicyRequest,
228        context: RequestContext,
229    ) -> Result<PolicyInfo> {
230        self.check_required(&request, &context).await?;
231        let for_securable_type = request
232            .for_securable_type
233            .as_deref()
234            .map(securable_type)
235            .transpose()?
236            .map(str::to_string);
237        let policy = PolicyInfo {
238            name: request.name,
239            on_securable_type: securable_type(&request.on_securable_type)?.to_string(),
240            on_securable_fullname: request.on_securable_fullname,
241            policy_type: request.policy_type,
242            comment: request.comment,
243            to_principals: request.to_principals,
244            except_principals: request.except_principals,
245            for_securable_type,
246            when_condition: request.when_condition,
247            match_columns: request.match_columns,
248            options: request_options(
249                request.row_filter.into_option(),
250                request.column_mask.into_option(),
251                request.deny.into_option(),
252                request.grant.into_option(),
253            )?,
254            ..Default::default()
255        };
256        validate_policy_shape(&policy)?;
257        tracing::Span::current().record("resource_name", &policy.name);
258
259        // Name uniqueness is scoped to the securable, not global.
260        let existing = self
261            .list(&ObjectLabel::PolicyInfo, None, None, None)
262            .await?
263            .0;
264        let clashes = existing.into_iter().any(|r| {
265            let Ok(p): std::result::Result<PolicyInfo, _> = r.try_into() else {
266                return false;
267            };
268            matches_securable(&p, &policy.on_securable_type, &policy.on_securable_fullname)
269                && p.name == policy.name
270        });
271        if clashes {
272            return Err(crate::Error::invalid_argument(format!(
273                "policy '{}' already exists on {} '{}'",
274                policy.name, policy.on_securable_type, policy.on_securable_fullname
275            )));
276        }
277
278        Ok(self.create(policy.into()).await?.0.try_into()?)
279    }
280
281    #[tracing::instrument(skip(self, context), fields(resource_name))]
282    async fn delete_policy(
283        &self,
284        request: DeletePolicyRequest,
285        context: RequestContext,
286    ) -> Result<()> {
287        tracing::Span::current().record("resource_name", &request.name);
288        self.check_required(&request, &context).await?;
289        let ident = find_policy_ident(
290            self,
291            securable_type(&request.on_securable_type)?,
292            &request.on_securable_fullname,
293            &request.name,
294        )
295        .await?;
296        Ok(self.delete(&ident).await?)
297    }
298
299    #[tracing::instrument(skip(self, context), fields(resource_name))]
300    async fn get_policy(
301        &self,
302        request: GetPolicyRequest,
303        context: RequestContext,
304    ) -> Result<PolicyInfo> {
305        tracing::Span::current().record("resource_name", &request.name);
306        self.check_required(&request, &context).await?;
307        let ident = find_policy_ident(
308            self,
309            securable_type(&request.on_securable_type)?,
310            &request.on_securable_fullname,
311            &request.name,
312        )
313        .await?;
314        Ok(self.get(&ident).await?.0.try_into()?)
315    }
316
317    #[tracing::instrument(skip(self, context))]
318    async fn list_policies(
319        &self,
320        request: ListPoliciesRequest,
321        context: RequestContext,
322    ) -> Result<ListPoliciesResponse> {
323        self.check_required(&request, &context).await?;
324        let on_securable_type = securable_type(&request.on_securable_type)?;
325
326        let (all, _) = self
327            .list(&ObjectLabel::PolicyInfo, None, None, None)
328            .await?;
329        let mut policies: Vec<PolicyInfo> = all.into_iter().map(|r| r.try_into()).try_collect()?;
330
331        if request.include_inherited.unwrap_or(false) {
332            let chain = securable_chain(on_securable_type, &request.on_securable_fullname);
333            policies.retain(|p| chain.iter().any(|(t, name)| matches_securable(p, t, name)));
334        } else {
335            policies.retain(|p| {
336                matches_securable(p, on_securable_type, &request.on_securable_fullname)
337            });
338        }
339
340        let mut resources: Vec<Resource> = policies.into_iter().map(Resource::PolicyInfo).collect();
341        process_resources(self, &context, &Permission::Read, &mut resources).await?;
342
343        let max_results = super::page_size(request.max_results)?.unwrap_or(resources.len());
344        let offset = request
345            .page_token
346            .as_deref()
347            .and_then(|t| t.parse::<usize>().ok())
348            .unwrap_or(0);
349        let next_page_token =
350            (offset + max_results < resources.len()).then(|| (offset + max_results).to_string());
351        let page = resources
352            .into_iter()
353            .skip(offset)
354            .take(max_results)
355            .map(|r| r.try_into())
356            .try_collect()?;
357
358        Ok(ListPoliciesResponse {
359            policies: page,
360            next_page_token,
361            ..Default::default()
362        })
363    }
364
365    #[tracing::instrument(skip(self, context), fields(resource_name))]
366    async fn update_policy(
367        &self,
368        mut request: UpdatePolicyRequest,
369        context: RequestContext,
370    ) -> Result<PolicyInfo> {
371        tracing::Span::current().record("resource_name", &request.name);
372        self.check_required(&request, &context).await?;
373        let ident = find_policy_ident(
374            self,
375            securable_type(&request.on_securable_type)?,
376            &request.on_securable_fullname,
377            &request.name,
378        )
379        .await?;
380        if let Some(t) = request.for_securable_type.as_deref() {
381            request.for_securable_type = Some(securable_type(t)?.to_string());
382        }
383        let mut policy: PolicyInfo = self.get(&ident).await?.0.try_into()?;
384        apply_update(&mut policy, request)?;
385        validate_policy_shape(&policy)?;
386        Ok(self.update(&ident, policy.into()).await?.0.try_into()?)
387    }
388}
389
390/// Resolve the [`ResourceIdent`] of the policy matching `name` on the given securable.
391///
392/// Policies are stored as flat objects keyed by their own `name`, so this scans all
393/// `PolicyInfo` objects rather than resolving via a hierarchical `ResourceName`.
394async fn find_policy_ident<S: ResourceStore>(
395    store: &S,
396    on_securable_type: &str,
397    on_securable_fullname: &str,
398    name: &str,
399) -> Result<ResourceIdent> {
400    let (all, _) = store
401        .list(&ObjectLabel::PolicyInfo, None, None, None)
402        .await?;
403    all.into_iter()
404        .find_map(|r| {
405            let p: PolicyInfo = r.try_into().ok()?;
406            (matches_securable(&p, on_securable_type, on_securable_fullname) && p.name == name)
407                .then(|| ResourceIdent::policy_info(ResourceName::new([name])))
408        })
409        .ok_or(crate::Error::NotFound)
410}
411
412impl SecuredAction for CreatePolicyRequest {
413    fn resource(&self) -> ResourceIdent {
414        securable_ident(&self.on_securable_type, &self.on_securable_fullname)
415    }
416
417    fn permission(&self) -> &'static Permission {
418        &Permission::Manage
419    }
420}
421
422impl SecuredAction for ListPoliciesRequest {
423    fn resource(&self) -> ResourceIdent {
424        securable_ident(&self.on_securable_type, &self.on_securable_fullname)
425    }
426
427    fn permission(&self) -> &'static Permission {
428        &Permission::Read
429    }
430}
431
432impl SecuredAction for GetPolicyRequest {
433    fn resource(&self) -> ResourceIdent {
434        securable_ident(&self.on_securable_type, &self.on_securable_fullname)
435    }
436
437    fn permission(&self) -> &'static Permission {
438        &Permission::Read
439    }
440}
441
442impl SecuredAction for UpdatePolicyRequest {
443    fn resource(&self) -> ResourceIdent {
444        securable_ident(&self.on_securable_type, &self.on_securable_fullname)
445    }
446
447    fn permission(&self) -> &'static Permission {
448        &Permission::Manage
449    }
450}
451
452impl SecuredAction for DeletePolicyRequest {
453    fn resource(&self) -> ResourceIdent {
454        securable_ident(&self.on_securable_type, &self.on_securable_fullname)
455    }
456
457    fn permission(&self) -> &'static Permission {
458        &Permission::Manage
459    }
460}
461
462/// Authorization for policy management is checked against the securable the policy is
463/// defined on, since defining a row-filter/column-mask is a governance action over that
464/// securable, not over the policy object itself.
465fn securable_ident(on_securable_type: &str, on_securable_fullname: &str) -> ResourceIdent {
466    let name = ResourceName::from_naive_str_split(on_securable_fullname);
467    match securable_type(on_securable_type) {
468        Ok("CATALOG") => ResourceIdent::catalog(name),
469        Ok("SCHEMA") => ResourceIdent::schema(name),
470        Ok("TABLE") => ResourceIdent::table(name),
471        _ => ResourceIdent::catalog(ResourceRef::Undefined),
472    }
473}
474
475#[cfg(test)]
476mod tests {
477    use std::sync::Arc;
478
479    use unitycatalog_common::models::catalogs::v1::Catalog;
480    use unitycatalog_common::models::schemas::v1::Schema;
481    use unitycatalog_common::models::tables::v1::{DataSourceFormat, Table, TableType};
482    use unitycatalog_common::services::encryption::{EnvelopeEncryptor, LocalKeyProvider};
483
484    use super::*;
485    use crate::memory::InMemoryResourceStore;
486    use crate::policy::ConstantPolicy;
487    use crate::services::ServerHandler;
488
489    async fn handler() -> ServerHandler<RequestContext> {
490        let encryptor =
491            EnvelopeEncryptor::local(LocalKeyProvider::single("test", vec![0x42; 32]).unwrap());
492        let store = Arc::new(InMemoryResourceStore::new(encryptor));
493        store
494            .create(
495                Catalog {
496                    name: "cat".to_string(),
497                    ..Default::default()
498                }
499                .into(),
500            )
501            .await
502            .unwrap();
503        store
504            .create(
505                Schema {
506                    name: "sch".to_string(),
507                    catalog_name: "cat".to_string(),
508                    ..Default::default()
509                }
510                .into(),
511            )
512            .await
513            .unwrap();
514        store
515            .create(
516                Table {
517                    name: "tbl".to_string(),
518                    catalog_name: "cat".to_string(),
519                    schema_name: "sch".to_string(),
520                    full_name: "cat.sch.tbl".to_string(),
521                    table_type: TableType::Managed.into(),
522                    data_source_format: DataSourceFormat::Delta.into(),
523                    ..Default::default()
524                }
525                .into(),
526            )
527            .await
528            .unwrap();
529        let policy: Arc<dyn Policy<RequestContext>> = Arc::new(ConstantPolicy::default());
530        ServerHandler::try_new_tokio(policy, store).unwrap()
531    }
532
533    fn ctx() -> RequestContext {
534        RequestContext {
535            recipient: crate::policy::Principal::anonymous(),
536        }
537    }
538
539    fn row_filter_request(
540        name: &str,
541        on_securable_type: &str,
542        on_securable_fullname: &str,
543    ) -> CreatePolicyRequest {
544        CreatePolicyRequest {
545            name: name.to_string(),
546            on_securable_type: on_securable_type.to_string(),
547            on_securable_fullname: on_securable_fullname.to_string(),
548            policy_type: PolicyType::RowFilter.into(),
549            to_principals: vec!["group:analysts".to_string()],
550            row_filter: Some(FunctionRef {
551                function_name: "cat.sch.my_filter".to_string(),
552                ..Default::default()
553            })
554            .into(),
555            ..Default::default()
556        }
557    }
558
559    fn update_request(name: &str) -> UpdatePolicyRequest {
560        UpdatePolicyRequest {
561            on_securable_type: "TABLE".to_string(),
562            on_securable_fullname: "cat.sch.tbl".to_string(),
563            name: name.to_string(),
564            ..Default::default()
565        }
566    }
567
568    #[tokio::test]
569    async fn policy_crud_round_trip() {
570        let h = handler().await;
571
572        let created = h
573            .create_policy(row_filter_request("p1", "TABLE", "cat.sch.tbl"), ctx())
574            .await
575            .unwrap();
576        assert_eq!(created.name, "p1");
577        assert_eq!(created.on_securable_fullname, "cat.sch.tbl");
578
579        let fetched = h
580            .get_policy(
581                GetPolicyRequest {
582                    on_securable_type: "table".to_string(),
583                    on_securable_fullname: "cat.sch.tbl".to_string(),
584                    name: "p1".to_string(),
585                    ..Default::default()
586                },
587                ctx(),
588            )
589            .await
590            .unwrap();
591        assert_eq!(fetched.name, "p1");
592
593        let listed = h
594            .list_policies(
595                ListPoliciesRequest {
596                    on_securable_type: "table".to_string(),
597                    on_securable_fullname: "cat.sch.tbl".to_string(),
598                    include_inherited: None,
599                    max_results: None,
600                    page_token: None,
601                    ..Default::default()
602                },
603                ctx(),
604            )
605            .await
606            .unwrap();
607        assert_eq!(listed.policies.len(), 1);
608
609        let updated = h
610            .update_policy(
611                UpdatePolicyRequest {
612                    comment: Some("updated".to_string()),
613                    ..update_request("p1")
614                },
615                ctx(),
616            )
617            .await
618            .unwrap();
619        assert_eq!(updated.comment.as_deref(), Some("updated"));
620
621        h.delete_policy(
622            DeletePolicyRequest {
623                on_securable_type: "table".to_string(),
624                on_securable_fullname: "cat.sch.tbl".to_string(),
625                name: "p1".to_string(),
626                ..Default::default()
627            },
628            ctx(),
629        )
630        .await
631        .unwrap();
632
633        let missing = h
634            .get_policy(
635                GetPolicyRequest {
636                    on_securable_type: "table".to_string(),
637                    on_securable_fullname: "cat.sch.tbl".to_string(),
638                    name: "p1".to_string(),
639                    ..Default::default()
640                },
641                ctx(),
642            )
643            .await;
644        assert!(missing.is_err(), "expected NotFound after delete");
645    }
646
647    #[tokio::test]
648    async fn list_include_inherited_unions_ancestor_chain() {
649        let h = handler().await;
650
651        h.create_policy(
652            row_filter_request("catalog_policy", "CATALOG", "cat"),
653            ctx(),
654        )
655        .await
656        .unwrap();
657        h.create_policy(
658            row_filter_request("schema_policy", "SCHEMA", "cat.sch"),
659            ctx(),
660        )
661        .await
662        .unwrap();
663        h.create_policy(
664            row_filter_request("table_policy", "TABLE", "cat.sch.tbl"),
665            ctx(),
666        )
667        .await
668        .unwrap();
669
670        let without_inheritance = h
671            .list_policies(
672                ListPoliciesRequest {
673                    on_securable_type: "table".to_string(),
674                    on_securable_fullname: "cat.sch.tbl".to_string(),
675                    include_inherited: None,
676                    max_results: None,
677                    page_token: None,
678                    ..Default::default()
679                },
680                ctx(),
681            )
682            .await
683            .unwrap();
684        assert_eq!(without_inheritance.policies.len(), 1);
685        assert_eq!(without_inheritance.policies[0].name, "table_policy");
686
687        let with_inheritance = h
688            .list_policies(
689                ListPoliciesRequest {
690                    on_securable_type: "table".to_string(),
691                    on_securable_fullname: "cat.sch.tbl".to_string(),
692                    include_inherited: Some(true),
693                    max_results: None,
694                    page_token: None,
695                    ..Default::default()
696                },
697                ctx(),
698            )
699            .await
700            .unwrap();
701        let mut names: Vec<_> = with_inheritance
702            .policies
703            .iter()
704            .map(|p| p.name.as_str())
705            .collect();
706        names.sort();
707        assert_eq!(
708            names,
709            vec!["catalog_policy", "schema_policy", "table_policy"]
710        );
711        // Each entry still carries its own securable, so the caller can see where it came from.
712        let by_name = |n: &str| {
713            with_inheritance
714                .policies
715                .iter()
716                .find(|p| p.name == n)
717                .unwrap()
718        };
719        assert_eq!(by_name("catalog_policy").on_securable_fullname, "cat");
720        assert_eq!(by_name("schema_policy").on_securable_fullname, "cat.sch");
721        assert_eq!(by_name("table_policy").on_securable_fullname, "cat.sch.tbl");
722    }
723
724    #[tokio::test]
725    async fn create_rejects_duplicate_name_on_same_securable() {
726        let h = handler().await;
727        h.create_policy(row_filter_request("p1", "TABLE", "cat.sch.tbl"), ctx())
728            .await
729            .unwrap();
730
731        let result = h
732            .create_policy(row_filter_request("p1", "TABLE", "cat.sch.tbl"), ctx())
733            .await;
734        assert!(
735            result.is_err(),
736            "duplicate policy name on same securable must be rejected"
737        );
738    }
739
740    #[tokio::test]
741    async fn create_rejects_column_mask_without_match_columns() {
742        let h = handler().await;
743        let request = CreatePolicyRequest {
744            policy_type: PolicyType::ColumnMask.into(),
745            row_filter: None.into(),
746            column_mask: Some(FunctionRef {
747                function_name: "cat.sch.my_mask".to_string(),
748                on_column: Some("ssn".to_string()),
749                ..Default::default()
750            })
751            .into(),
752            ..row_filter_request("bad_mask", "TABLE", "cat.sch.tbl")
753        };
754        let result = h.create_policy(request, ctx()).await;
755        assert!(
756            result.is_err(),
757            "column mask without match_columns must be rejected"
758        );
759    }
760
761    #[tokio::test]
762    async fn get_unknown_policy_is_not_found() {
763        let h = handler().await;
764        let result = h
765            .get_policy(
766                GetPolicyRequest {
767                    on_securable_type: "table".to_string(),
768                    on_securable_fullname: "cat.sch.tbl".to_string(),
769                    name: "does-not-exist".to_string(),
770                    ..Default::default()
771                },
772                ctx(),
773            )
774            .await;
775        assert!(result.is_err(), "unknown policy must be NotFound");
776    }
777
778    #[tokio::test]
779    async fn update_mask_clears_listed_fields_and_keeps_others() {
780        let h = handler().await;
781        let mut request = row_filter_request("p1", "TABLE", "cat.sch.tbl");
782        request.comment = Some("original".to_string());
783        request.when_condition = Some("hasTag('pii')".to_string());
784        h.create_policy(request, ctx()).await.unwrap();
785
786        // Without a mask only the fields set in the request change.
787        let updated = h
788            .update_policy(
789                UpdatePolicyRequest {
790                    comment: Some("changed".to_string()),
791                    ..update_request("p1")
792                },
793                ctx(),
794            )
795            .await
796            .unwrap();
797        assert_eq!(updated.comment.as_deref(), Some("changed"));
798        assert_eq!(updated.when_condition.as_deref(), Some("hasTag('pii')"));
799        assert_eq!(updated.to_principals, vec!["group:analysts".to_string()]);
800
801        // A field named in the mask but unset in the request is cleared.
802        let cleared = h
803            .update_policy(
804                UpdatePolicyRequest {
805                    update_mask: Some("when_condition".to_string()),
806                    ..update_request("p1")
807                },
808                ctx(),
809            )
810            .await
811            .unwrap();
812        assert_eq!(cleared.when_condition, None);
813        assert_eq!(cleared.comment.as_deref(), Some("changed"));
814
815        let unknown = h
816            .update_policy(
817                UpdatePolicyRequest {
818                    update_mask: Some("name".to_string()),
819                    ..update_request("p1")
820                },
821                ctx(),
822            )
823            .await;
824        assert!(
825            unknown.is_err(),
826            "masking a non-updatable field must be rejected"
827        );
828    }
829
830    #[tokio::test]
831    async fn deny_policy_requires_deny_options() {
832        let h = handler().await;
833        let missing = CreatePolicyRequest {
834            policy_type: PolicyType::Deny.into(),
835            row_filter: None.into(),
836            ..row_filter_request("deny_missing", "CATALOG", "cat")
837        };
838        assert!(h.create_policy(missing, ctx()).await.is_err());
839
840        let created = h
841            .create_policy(
842                CreatePolicyRequest {
843                    policy_type: PolicyType::Deny.into(),
844                    row_filter: None.into(),
845                    deny: Some(PrivilegeOptions {
846                        privileges: vec!["SELECT".to_string()],
847                        ..Default::default()
848                    })
849                    .into(),
850                    ..row_filter_request("deny_select", "catalog", "cat")
851                },
852                ctx(),
853            )
854            .await
855            .unwrap();
856        // The securable type is stored in its canonical upper-case form.
857        assert_eq!(created.on_securable_type, "CATALOG");
858    }
859}
unitycatalog_server/api/policies.rs

unitycatalog_server/api/
policies.rs
