mirror of
https://github.com/danswer-ai/danswer.git
synced 2025-03-27 10:13:05 +01:00
122 lines
3.8 KiB
Python
122 lines
3.8 KiB
Python
from dataclasses import dataclass
|
|
|
|
from onyx.access.utils import prefix_external_group
|
|
from onyx.access.utils import prefix_user_email
|
|
from onyx.access.utils import prefix_user_group
|
|
from onyx.configs.constants import PUBLIC_DOC_PAT
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ExternalAccess:
|
|
# Emails of external users with access to the doc externally
|
|
external_user_emails: set[str]
|
|
# Names or external IDs of groups with access to the doc
|
|
external_user_group_ids: set[str]
|
|
# Whether the document is public in the external system or Onyx
|
|
is_public: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DocExternalAccess:
|
|
"""
|
|
This is just a class to wrap the external access and the document ID
|
|
together. It's used for syncing document permissions to Redis.
|
|
"""
|
|
|
|
external_access: ExternalAccess
|
|
# The document ID
|
|
doc_id: str
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"external_access": {
|
|
"external_user_emails": list(self.external_access.external_user_emails),
|
|
"external_user_group_ids": list(
|
|
self.external_access.external_user_group_ids
|
|
),
|
|
"is_public": self.external_access.is_public,
|
|
},
|
|
"doc_id": self.doc_id,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "DocExternalAccess":
|
|
external_access = ExternalAccess(
|
|
external_user_emails=set(
|
|
data["external_access"].get("external_user_emails", [])
|
|
),
|
|
external_user_group_ids=set(
|
|
data["external_access"].get("external_user_group_ids", [])
|
|
),
|
|
is_public=data["external_access"]["is_public"],
|
|
)
|
|
return cls(
|
|
external_access=external_access,
|
|
doc_id=data["doc_id"],
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DocumentAccess(ExternalAccess):
|
|
# User emails for Onyx users, None indicates admin
|
|
user_emails: set[str | None]
|
|
# Names of user groups associated with this document
|
|
user_groups: set[str]
|
|
|
|
def to_acl(self) -> set[str]:
|
|
return set(
|
|
[
|
|
prefix_user_email(user_email)
|
|
for user_email in self.user_emails
|
|
if user_email
|
|
]
|
|
+ [prefix_user_group(group_name) for group_name in self.user_groups]
|
|
+ [
|
|
prefix_user_email(user_email)
|
|
for user_email in self.external_user_emails
|
|
]
|
|
+ [
|
|
# The group names are already prefixed by the source type
|
|
# This adds an additional prefix of "external_group:"
|
|
prefix_external_group(group_name)
|
|
for group_name in self.external_user_group_ids
|
|
]
|
|
+ ([PUBLIC_DOC_PAT] if self.is_public else [])
|
|
)
|
|
|
|
@classmethod
|
|
def build(
|
|
cls,
|
|
user_emails: list[str | None],
|
|
user_groups: list[str],
|
|
external_user_emails: list[str],
|
|
external_user_group_ids: list[str],
|
|
is_public: bool,
|
|
) -> "DocumentAccess":
|
|
return cls(
|
|
external_user_emails={
|
|
prefix_user_email(external_email)
|
|
for external_email in external_user_emails
|
|
},
|
|
external_user_group_ids={
|
|
prefix_external_group(external_group_id)
|
|
for external_group_id in external_user_group_ids
|
|
},
|
|
user_emails={
|
|
prefix_user_email(user_email)
|
|
for user_email in user_emails
|
|
if user_email
|
|
},
|
|
user_groups=set(user_groups),
|
|
is_public=is_public,
|
|
)
|
|
|
|
|
|
default_public_access = DocumentAccess(
|
|
external_user_emails=set(),
|
|
external_user_group_ids=set(),
|
|
user_emails=set(),
|
|
user_groups=set(),
|
|
is_public=True,
|
|
)
|