mirror of
https://github.com/danswer-ai/danswer.git
synced 2025-03-27 10:13:05 +01:00
109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
from sqlalchemy.orm import Session
|
|
|
|
from onyx.access.models import DocumentAccess
|
|
from onyx.access.utils import prefix_user_email
|
|
from onyx.configs.constants import PUBLIC_DOC_PAT
|
|
from onyx.db.document import get_access_info_for_document
|
|
from onyx.db.document import get_access_info_for_documents
|
|
from onyx.db.models import User
|
|
from onyx.utils.variable_functionality import fetch_versioned_implementation
|
|
|
|
|
|
def _get_access_for_document(
|
|
document_id: str,
|
|
db_session: Session,
|
|
) -> DocumentAccess:
|
|
info = get_access_info_for_document(
|
|
db_session=db_session,
|
|
document_id=document_id,
|
|
)
|
|
|
|
return DocumentAccess.build(
|
|
user_emails=info[1] if info and info[1] else [],
|
|
user_groups=[],
|
|
external_user_emails=[],
|
|
external_user_group_ids=[],
|
|
is_public=info[2] if info else False,
|
|
)
|
|
|
|
|
|
def get_access_for_document(
|
|
document_id: str,
|
|
db_session: Session,
|
|
) -> DocumentAccess:
|
|
versioned_get_access_for_document_fn = fetch_versioned_implementation(
|
|
"onyx.access.access", "_get_access_for_document"
|
|
)
|
|
return versioned_get_access_for_document_fn(document_id, db_session) # type: ignore
|
|
|
|
|
|
def get_null_document_access() -> DocumentAccess:
|
|
return DocumentAccess(
|
|
user_emails=set(),
|
|
user_groups=set(),
|
|
is_public=False,
|
|
external_user_emails=set(),
|
|
external_user_group_ids=set(),
|
|
)
|
|
|
|
|
|
def _get_access_for_documents(
|
|
document_ids: list[str],
|
|
db_session: Session,
|
|
) -> dict[str, DocumentAccess]:
|
|
document_access_info = get_access_info_for_documents(
|
|
db_session=db_session,
|
|
document_ids=document_ids,
|
|
)
|
|
doc_access = {
|
|
document_id: DocumentAccess(
|
|
user_emails=set([email for email in user_emails if email]),
|
|
# MIT version will wipe all groups and external groups on update
|
|
user_groups=set(),
|
|
is_public=is_public,
|
|
external_user_emails=set(),
|
|
external_user_group_ids=set(),
|
|
)
|
|
for document_id, user_emails, is_public in document_access_info
|
|
}
|
|
|
|
# Sometimes the document has not be indexed by the indexing job yet, in those cases
|
|
# the document does not exist and so we use least permissive. Specifically the EE version
|
|
# checks the MIT version permissions and creates a superset. This ensures that this flow
|
|
# does not fail even if the Document has not yet been indexed.
|
|
for doc_id in document_ids:
|
|
if doc_id not in doc_access:
|
|
doc_access[doc_id] = get_null_document_access()
|
|
return doc_access
|
|
|
|
|
|
def get_access_for_documents(
|
|
document_ids: list[str],
|
|
db_session: Session,
|
|
) -> dict[str, DocumentAccess]:
|
|
"""Fetches all access information for the given documents."""
|
|
versioned_get_access_for_documents_fn = fetch_versioned_implementation(
|
|
"onyx.access.access", "_get_access_for_documents"
|
|
)
|
|
return versioned_get_access_for_documents_fn(
|
|
document_ids, db_session
|
|
) # type: ignore
|
|
|
|
|
|
def _get_acl_for_user(user: User | None, db_session: Session) -> set[str]:
|
|
"""Returns a list of ACL entries that the user has access to. This is meant to be
|
|
used downstream to filter out documents that the user does not have access to. The
|
|
user should have access to a document if at least one entry in the document's ACL
|
|
matches one entry in the returned set.
|
|
"""
|
|
if user:
|
|
return {prefix_user_email(user.email), PUBLIC_DOC_PAT}
|
|
return {PUBLIC_DOC_PAT}
|
|
|
|
|
|
def get_acl_for_user(user: User | None, db_session: Session | None = None) -> set[str]:
|
|
versioned_acl_for_user_fn = fetch_versioned_implementation(
|
|
"onyx.access.access", "_get_acl_for_user"
|
|
)
|
|
return versioned_acl_for_user_fn(user, db_session) # type: ignore
|