#
# Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending
#
"""This module supports the creation of access control lists (ACLs) for the results of Persistent Queries.
For more information on ACLs and filter generators, see the Deephaven documentation
https://deephaven.io/enterprise/docs/sys-admin/permissions/acls/#persistent-query-acls.
"""
from __future__ import annotations
from typing import Any, TypeVar, Union, cast
import jpy
from deephaven import DHError
from deephaven._wrapper import JObjectWrapper, unwrap
from deephaven.jcompat import j_array_list, to_sequence
from deephaven.table import PartitionedTable, RollupTable, Table, TreeTable
from deephaven_enterprise.pivot.pivot_table import PivotTable
# Union of all supported table types for apply_to
TableType = TypeVar(
"TableType",
bound=Union[Table, RollupTable, TreeTable, PivotTable, PartitionedTable],
)
_JEdgeAclProvider = jpy.get_type("io.deephaven.enterprise.acl.EdgeAclProvider")
_JEdgeAclProviderBuilder: Any = jpy.get_type(
"io.deephaven.enterprise.acl.EdgeAclProvider$Builder"
)
JAclFilterGenerator: Any = jpy.get_type(
"io.deephaven.enterprise.acl.AclFilterGenerator"
)
[docs]
class EdgeAclProvider:
"""An EdgeAclProvider is used to provide access control for the results of Persistent Queries based on the
requesting user at request time.
Note that it should not be instantiated directly, but rather through the :meth:`EdgeAclProviderBuilder.build` method.
"""
@staticmethod
[docs]
def builder() -> EdgeAclProviderBuilder:
"""Creates a builder to construct an EdgeAclProvider
Returns:
EdgeAclProviderBuilder
"""
return EdgeAclProviderBuilder(_JEdgeAclProvider.builder())
@staticmethod
[docs]
def apply_full_access(table: TableType) -> TableType:
"""Applies a full-access ("allusers") Edge ACL to the given table.
This is a convenience for building a provider with a single full-access row ACL for the ``allusers`` group and
applying it::
EdgeAclProvider.builder().row_acl("allusers", acl_generator.full_access()).build().apply_to(table)
The resulting table is not transformed for any user, but is marked as ACL-bearing so that it may be exported
once ACLs have been applied somewhere in the query.
Args:
table (TableType): the object to attach to, must be a Table, RollupTable, TreeTable, PivotTable, or PartitionedTable
Returns:
a new table with a full-access ACL attached
"""
return EdgeAclProvider(_JEdgeAclProvider.fullAccess()).apply_to(table)
@staticmethod
[docs]
def apply_full_access_for_current_user(table: TableType) -> TableType:
"""Applies a full-access Edge ACL for the current user to the given table.
Unlike :meth:`apply_full_access`, which grants access to the ``allusers`` group, this grants full access only to
the user in the current authorization context (a user is always a member of a group named for themselves). Use
it to mark a result computed in a viewer's context as exportable to that viewer alone. The caller is responsible
for ensuring the data is already sanitized for that user.
Args:
table (TableType): the object to attach to, must be a Table, RollupTable, TreeTable, PivotTable, or PartitionedTable
Returns:
a new table with a current-user full-access ACL attached
"""
return EdgeAclProvider(_JEdgeAclProvider.fullAccessForCurrentUser()).apply_to(
table
)
@staticmethod
[docs]
def add_object_acl(group: str, value: Union[jpy.JType, Any]) -> None:
"""Adds visibility for the specified object to the specified group.
Args:
group (str): the group to add visibility for
value (Union[jpy.JType, Any]): the object to add visibility for
"""
if isinstance(value, JObjectWrapper):
_JEdgeAclProvider.addAcl(group, unwrap(value))
else:
_JEdgeAclProvider.addPyObjectAcl(group, value)
@staticmethod
[docs]
def add_full_access(value: Union[jpy.JType, Any]) -> None:
"""Adds visibility for the specified object to all users and groups.
Args:
value (Union[jpy.JType, Any]): the object to add visibility for
"""
if isinstance(value, JObjectWrapper):
_JEdgeAclProvider.addFullAccess(unwrap(value))
else:
_JEdgeAclProvider.addPyFullAccess(value)
def __init__(self, j_provider: jpy.JType):
self.j_provider = jpy.cast(j_provider, _JEdgeAclProvider)
if self.j_provider is None:
raise DHError(
message="j_provider type is not io.deephaven.enterprise.acl.EdgeAclProvider"
)
[docs]
def apply_to(self, table: TableType) -> TableType:
"""Attaches this ACL Provider to a table
Args:
table (TableType): the object to attach to, must be a Table, RollupTable, TreeTable, PivotTable, or PartitionedTable
Returns:
a new table with the ACLs attached
"""
if isinstance(table, RollupTable):
return cast(
TableType,
RollupTable(
self.j_provider.applyTo(table.j_object),
aggs=table.aggs,
by=table.by,
include_constituents=table.include_constituents,
),
)
elif isinstance(table, TreeTable):
return cast(
TableType,
TreeTable(
self.j_provider.applyTo(table.j_object),
id_col=table.id_col,
parent_col=table.parent_col,
),
)
elif isinstance(table, PivotTable):
return cast(TableType, PivotTable(self.j_provider.applyTo(table.j_object)))
elif isinstance(table, PartitionedTable):
return cast(
TableType,
PartitionedTable(
j_partitioned_table=self.j_provider.applyTo(table.j_object)
),
)
return cast(TableType, Table(self.j_provider.applyTo(table.j_object)))
[docs]
class EdgeAclProviderBuilder(JObjectWrapper):
"""A builder for constructing an EdgeAclProvider for access control on the result tables of Persistent Queries.
Note that it should not be instantiated directly, but rather through the static :meth:`EdgeAclProvider.builder`.
"""
j_object_type = _JEdgeAclProviderBuilder
@property
def j_object(self) -> jpy.JType:
return self.j_builder
def __init__(self, j_builder: _JEdgeAclProviderBuilder):
self.j_builder = j_builder
[docs]
def row_acl(self, group: str, acl: JAclFilterGenerator) -> EdgeAclProviderBuilder:
"""Adds a Row ACL for the specified group to the table with a filter generator. The filter generator is created
by calling one of the factory functions in :mod:`.acl_generator`. The Row Acl controls access to rows of a table.
Args:
group (str): the group the ACL applies to
acl (JAclFilterGenerator): the generator that will produce a filter for a given user
Returns:
self
"""
self.j_builder.rowAcl(group, acl)
return self
[docs]
def column_acl(
self, group: str, columns: Union[str, list[str]], acl: JAclFilterGenerator
) -> EdgeAclProviderBuilder:
"""Adds a Column ACL for the specified group and column(s) to the table with a filter generator. The filter
generator is created by calling one of the factory functions in :mod:`.acl_generator`. The Column ACL controls
access to specific values of columns of a table.
Args:
group (str): the group the ACL applies to
columns (Union[str, list[str]]): the column(s) the acl applies to
acl (JAclFilterGenerator): the generator that will produce a filter for a given user
Returns:
self
"""
columns = list(to_sequence(columns))
self.j_builder.columnAcl(group, j_array_list(columns), acl)
return self
[docs]
def build(self) -> EdgeAclProvider:
"""Constructs the completed :class:`EdgeAclProvider` object from the current state of this builder.
Returns:
EdgeAclProvider
"""
return EdgeAclProvider(self.j_builder.build())