khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20241211) Figuring out the inherited permissions system. Stuck at directory management with permissions handling.

Browse Source
This commit is contained in:
khushal
2024-12-11 19:00:14 +05:30
parent 448b183662
commit 75d269c9a5
4 changed files with 442 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files
models/behaviour/file/file_object.py
+344 -209
View File
@@ -31,17 +31,14 @@ import io
# To make sibling directories accessible for imports:
import sys
from sqlalchemy.orm.collections import collection
from utils_v2.mail.mail_parser_v2 import parse_addr
sys.path.append(".")
sys.path.append("..")
# My async utils:
from utils_v2.string import json
from utils_v2.date_time import date_time
from utils_v2.system import files
from utils_v2.security.hash import Hasher
from utils_v2.database.async_mysql_v2 import AsyncMySQL
from utils_v2.database.async_mongo_v2 import AsyncMongo, AsyncMongoStorage
@@ -53,7 +50,8 @@ from models.data.core.user import CoreUserInfoModel
from models.data.core.file_object import (
CoreFileObjectInfoModel,
CoreFileObjectSharingModel,
CoreFileObjectPermissionsModel
CoreFileObjectPermissionsModel,
CoreFileObjectAccessResponseModel
)
# To work with MongoDB:
@@ -65,8 +63,8 @@ from typing import Any, Literal, List
# To make deep copies:
import copy
# To make deep-copies:
import copy
# For debugging:
from icecream import IceCreamDebugger
# *****************************************************************************************************************
@@ -111,75 +109,39 @@ class FileObjectManagementModel:
# Define class-level variables:
FILE_OBJECTS_COLLECTION = "_fileObjects"
def __init__(
self,
debug = True,
debug_prefix = "File Objs. | ",
):
# Debugging:
self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
if not debug: self._printer.disable()
def enable_debug(self):
self._printer.enable()
def disable_debug(self):
self._printer.disable()
# ┓┏ ┓
# ┣┫┏┓┃┏┓┏┓┏┓┏
# ┛┗┗ ┗┣┛┗ ┛ ┛
# ┛
def match_lists(
self,
list_a: list,
list_b: list
):
@staticmethod
def users_match(
user_p: CoreUserInfoModel,
user_r: CoreUserInfoModel,
ignore_null: bool = True
) -> bool:
"""
To match items in a List. The lists should have values that can be sorted. Mixed value dicts will raise
exceptions.
:param list_a: One of the lists to check.
:param list_b: The other list to check.
:return: True if they match, else False.
"""
# Start by assuming success:
are_matching = True
# If the lengths don't match,
# there is no way the lists are exactly the same:
if len(list_a) != len(list_b):
are_matching = False
return are_matching
# Sort the lists first for better comparison:
sorted_list_a = sorted(copy.deepcopy(list_a))
sorted_list_b = sorted(copy.deepcopy(list_b))
# Now loop through them and compare elements at the same index:
for index in range(len(sorted_list_a)):
av = sorted_list_a[index]
bv = sorted_list_b[index]
# If both of them are some sorts of lists:
if isinstance(av, (list, tuple, set)):
if not self.match_lists(list_a = list(av), list_b = list(bv)):
are_matching = False
break
# If both of them are dicts, we go recursive:
elif isinstance(av, dict):
if not self.match_dicts(dict_a = av, dict_b = bv):
are_matching = False
break
# Finally, we make direct comparison for things like strings, ints, etc.
else:
if av != bv:
are_matching = False
break
# Done here:
return are_matching
def match_dicts(
self,
dict_a: dict,
dict_b: dict
):
"""
To match items in a dict. Any lists in the dict should have values that can be sorted. Mixed value dicts will
raise exceptions.
:param dict_a: One of the dicts to check.
:param dict_b: The other dict to check.
To match if a user that is requesting a resource is the same as the user known to have access to the resource.
:param user_p: One of the dicts to check.
:param user_r: The other dict to check.
:param ignore_null: Whether to consider only non-null values, or all values.
:return: True if they match, else False.
"""
@@ -187,35 +149,21 @@ class FileObjectManagementModel:
are_matching = True
# Iterate through the required items:
for ak, av in dict_a.items():
for rk, rv in user_r.model_dump().items():
# Fetch the corresponding value from the other dict:
bv = dict_b.get(ak)
# Do not consider fields that are nulls if asked to ignore them:
if ignore_null and rv is None: continue
# If the types are themselves mismatched,
# no point in further comparison:
if not isinstance(av, type(bv)):
# Extract the corresponding value from the other user,
# and test it for being equal:
pv = getattr(user_p, rk, None)
if (
(not isinstance(rv, type(pv))) or
(rv != pv)
):
are_matching = False
break
# If both of them are some sorts of lists:
if isinstance(av, (list, tuple, set)):
if not self.match_lists(list_a = list(av), list_b = list(bv)):
are_matching = False
break
# If both of them are dicts, we go recursive:
elif isinstance(av, dict):
if not self.match_dicts(dict_a = av, dict_b = bv):
are_matching = False
break
# Finally, we make direct comparison for things like strings, ints, etc.
else:
if av != bv:
are_matching = False
break
# Done here:
return are_matching
@@ -227,48 +175,85 @@ class FileObjectManagementModel:
self,
mongo_conn: AsyncMongoStorage,
file_object_id: ObjectId | str
) -> bool:
) -> CoreFileObjectAccessResponseModel:
"""
To check whether, or not, a particular file's record exists in the database.
:param mongo_conn: The instance of the database connection to perform this action.
:param file_object_id: The id of the file to check.
:return: True if it exists, else False.
:return: A structured response where the existence of the file is noted in the 'result' field.
"""
# Run the query:
record = await mongo_conn.find_one(
collection = self.FILE_OBJECTS_COLLECTION,
filter = {"_id": ObjectId(file_object_id)},
projection = {"_id": True, "user": True}
)
# Create a response:
response = CoreFileObjectAccessResponseModel()
# Return the result:
if not record: return False
else: return True
try:
# Run the query:
record = await mongo_conn.find_one(
collection = self.FILE_OBJECTS_COLLECTION,
filter = {"_id": ObjectId(file_object_id)},
projection = {"_id": True, "user": True, "isDir": True},
raise_exception = True
)
# Note down the result:
if record:
response.result = True
response.success = True
response.message = "ok"
except Exception as exception:
response.exception = exception
response.message = str(exception)
response.success = False
response.result = None
# Done here:
return response
async def info(
self,
mongo_conn: AsyncMongoStorage,
file_object_id: ObjectId
) -> CoreFileObjectInfoModel | None:
file_object_id: ObjectId | str
) -> CoreFileObjectAccessResponseModel:
"""
To get the information about this file.
:param mongo_conn: The instance of the database connection to perform this action.
:param file_object_id: The id of the file to check.
:return: None if the file doesn't exist, else its summary.
:return: A structured response where the info of the file is noted in the 'result' field.
"""
# Run the query:
record = await mongo_conn.find_one(
collection = self.FILE_OBJECTS_COLLECTION,
filter = {"_id": ObjectId(file_object_id)}
)
# Create a response:
response = CoreFileObjectAccessResponseModel()
# return the results:
if not record: return None
else: return CoreFileObjectInfoModel(**record)
try:
# Run the query:
record = await mongo_conn.find_one(
collection = self.FILE_OBJECTS_COLLECTION,
filter = {"_id": ObjectId(file_object_id)},
raise_exception = True
)
# Note down the result:
response.success = True
if record:
response.result = CoreFileObjectInfoModel(**record)
response.message = "ok"
else:
response.message = "no such file object"
# If something goes wrong:
except Exception as exception:
response.exception = exception
response.message = str(exception)
response.success = False
response.result = None
# Done here:
return response
async def is_private(
self,
@@ -283,82 +268,88 @@ class FileObjectManagementModel:
:return: True if private, else False. None if it doesn't exist at all.
"""
# Run the query:
record = await mongo_conn.find_one(
collection = self.FILE_OBJECTS_COLLECTION,
filter = {"_id": ObjectId(file_object_id)},
projection = {"_id": False, "isPrivate": True}
)
# Create a response:
response = CoreFileObjectAccessResponseModel()
# Return the result:
if not record: return None
else: return record["isPrivate"]
try:
async def is_public(
self,
mongo_conn: AsyncMongoStorage,
file_object_id: ObjectId | str
) -> bool | None:
# Run the query:
record = await mongo_conn.find_one(
collection = self.FILE_OBJECTS_COLLECTION,
filter = {"_id": ObjectId(file_object_id)},
projection = {"_id": False, "isPrivate": True},
raise_exception = True
)
"""
A wrapper around the is_private, method that returns the opposite value.
:param mongo_conn: The instance of the database connection to perform this action.
:param file_object_id: The id of the file to check.
:return: True if public, else False. None if it doesn't exist at all.
"""
# Note down the result:
if record:
response.result = record["isPrivate"]
response.success = True
response.message = "ok"
# Call the existing function:
is_private = await self.is_private(
mongo_conn = mongo_conn,
file_object_id = file_object_id
)
# If something goes wrong:
except Exception as exception:
response.exception = exception
response.message = str(exception)
response.success = False
response.result = None
# Return the opposite result:
if is_private is None: return is_private
else: return not is_private
# Done here:
return response
async def is_owner(
def is_owner(
self,
mongo_conn: AsyncMongoStorage,
user_info: CoreUserInfoModel,
file_object_id: ObjectId | str,
) -> bool | None:
file_object_info: CoreFileObjectInfoModel,
ignore_null: bool = True
) -> CoreFileObjectAccessResponseModel:
"""
To check if a specific user is the owner of a specific file.
:param mongo_conn: The instance of the database connection to perform this action.
:param user_info: The details of the user who needs to have permissions to this file.
:param file_object_id: The id of the file to check.
:param file_object_info: The information about the file/dir. Fetch it from the 'info' method.
:param ignore_null: Whether to consider only non-null values, or all values.
:return: True if owner, else False. None if something goes wrong.
"""
# Parse the user conditions by ignoring the null values:
user_conditions = {k: v for k, v in user_info.model_dump().items() if v is not None}
# Create a response:
response = CoreFileObjectAccessResponseModel()
# Run the query:
record = await mongo_conn.find_one(
collection = self.FILE_OBJECTS_COLLECTION,
filter = mongo_conn.dict_to_dot_notation({
"_id": ObjectId(file_object_id),
"user": user_conditions,
}),
projection = {
"user": True
}
)
try:
# Return the result:
if record is None: return False
else: return True
# Test for a match:
if self.users_match(
user_p = file_object_info.user,
user_r = user_info,
ignore_null = ignore_null
):
response.result = True
response.message = "user is the owner of this resource"
else:
response.result = False
response.message = "user is not the owner of this resource"
response.success = True
async def has_permission(
# If something goes wrong:
except Exception as exception:
response.exception = exception
response.message = str(exception)
response.success = False
response.result = None
# Done here:
return response
def has_permission(
self,
mongo_conn: AsyncMongoStorage,
user_info: CoreUserInfoModel,
permission: Literal["read", "delete", "changePermissions"],
file_object_id: ObjectId | str = None,
check_ownership: bool = True
) -> bool | None:
file_object_info: CoreFileObjectInfoModel,
permission: Literal["read", "write", "delete", "changePermissions"],
ignore_null: bool = True
) -> CoreFileObjectAccessResponseModel:
"""
To check if a particular user has permissions to a given file obj. You may pass either an instance of the file's
@@ -366,40 +357,56 @@ class FileObjectManagementModel:
:param mongo_conn: The instance of the database connection to perform this action.
:param user_info: The details of the user who needs to have permissions to this file.
:param permission: The name of the permission that the said user must have on this file.
:param file_object_id: The id of the file to check.
:param check_ownership: Owners of files have all rights to their own files. If this field is set to True,
ownership will be tested, and if the user is the owner of the file, the permission is assumed to be granted.
If this field is set to False, only the sharing section of the file's record will be checked.
:param file_object_info: The information about the file/dir. Fetch it from the 'info' method.
:param ignore_null: Whether to consider only non-null values, or all values.
:return: True if the user has said permission, else False. None if something goes wrong.
"""
# Parse the user conditions by ignoring the null values:
user_conditions = {k: v for k, v in user_info.model_dump().items() if v is not None}
# If we have to test ownership:
if check_ownership:
is_owner = await self.is_owner(
mongo_conn = mongo_conn,
user_info = user_info,
file_object_id = file_object_id
# If this is a public file/dir,
# and the permission requested is 'read':
if (not file_object_info.isPrivate) and permission == "read":
return CoreFileObjectAccessResponseModel(
success = True,
message = "this resource is publicly available",
result = True,
exception = None
)
if is_owner: return True
# otherwise, we run the query to check for granted permissions:
record = await mongo_conn.find_one(
collection = self.FILE_OBJECTS_COLLECTION,
filter = mongo_conn.dict_to_dot_notation({
"_id": ObjectId(file_object_id),
"sharedWith.user": user_conditions,
}),
projection = {
"sharedWith.$": True
}
# The owner always has all permissions:
response = self.is_owner(
mongo_conn = mongo_conn,
user_info = user_info,
file_object_info = file_object_info,
ignore_null = ignore_null
)
if not response.success: return response
if response.result is True: return response
# Return the result:
if record is None: return False
else: return record["sharedWith"][0]["permissions"][permission]
# Note down the failure of the ownership test:
response.message = "this user does not have the requested permission over this resource"
# Since The person requesting this is not the owner,
# we check with the sharing details:
for sharing_data in file_object_info.sharedWith:
# We match the users.
# If they don't match, we move to the next user:
if not self.users_match(
user_p = sharing_data.user,
user_r = user_info,
ignore_null = ignore_null
): continue
# If we found a matching user,
# we check for the permission:
if getattr(sharing_data.permissions, permission, False):
response.result = True
response.success = True
response.message = "this user has the requested permission over this resource"
break
# Done here:
return response
# ┓ • •
# ┃ ┓┏╋┓┏┓┏┓
@@ -603,11 +610,120 @@ class FileObjectManagementModel:
# ┗┻┛┛ ┗┗┗┛┗┗┫
# ┛
async def make_dir(self): pass
async def make_dir(
self,
mongo_conn: AsyncMongoStorage,
user_info: CoreUserInfoModel,
dir_name: str,
dir_metadata: dict = None,
dir_tags: list = None,
parent_id: ObjectId | str = None
) -> bool:
async def upload_file(self): pass
# Create an instance of the directory's model:
dir_model = CoreFileObjectInfoModel(
_id = mongo_conn.generate_id(),
user = user_info,
isDir = True,
name = dir_name,
createTs = date_time.get_current_utc_date_time(as_string = False),
metadata = dir_metadata,
tags = dir_tags,
parentId = parent_id,
isPrivate = True
)
async def get_file_upload_stream(self): pass
print("DIRECTORY:", json.to_string(dir_model.model_dump(), default = str))
# Insert this document into the database:
inserted_id = await mongo_conn.insert_one(
collection = self.FILE_OBJECTS_COLLECTION,
document = dir_model.model_dump()
)
# Done here:
return True if inserted_id else False
async def upload_file(
self,
mongo_conn: AsyncMongoStorage,
file_info: CoreFileObjectInfoModel,
file_data: io.BytesIO | str
):
# Start by assuming failure:
file_uploaded = False
# Start a session:
async with await (await mongo_conn.client).start_session() as session:
# Define the transaction options:
options = {
# "read_concern": {"level": "snapshot"}, # ... Optional: ensures consistent reads.
# "write_concern": {"w": "majority"}, # ...... Ensures writes are acknowledged.
# "read_preference": "primary", # ............ Specify where to read from (e.g., primary).
}
# Start the transaction:
async with session.start_transaction(**options):
try:
# Check that we indeed have a file that we are uploading:
if file_info.isDir: raise ValueError("cannot data to upload directory object")
# Generate an ObjectId:
file_info.fileObjectId = mongo_conn.generate_id(as_str = False)
# Read the file's data into a BytesIO object:
if isinstance(file_data, str):
file_data = io.BytesIO(files.read_file(file_data, mode = "rb"))
file_data.seek(0)
# Hash the file's data:
hasher = Hasher()
hasher.update(file_data.getvalue())
file_info.hash = hasher.hexdigest()
# Now we upload the actual content of the file with the same id:
success = await mongo_conn.easy_upload(
source = file_data,
file_name = file_info.name,
file_metadata = None,
file_id = file_info.fileObjectId,
session = session,
raise_exception = True
)
# Safety check to ensure that the data was written:
if not success:
raise ValueError("file object's bytes weren't uploaded")
# Write the file object's info model now with the same id:
inserted_id = await mongo_conn.insert_one(
collection = self.FILE_OBJECTS_COLLECTION,
document = file_info.model_dump(),
session = session,
raise_exception = True
)
# Safety check to ensure that the info was written:
if inserted_id is None:
raise ValueError("file object's info wasn't inserted")
# If we've reached this far:
file_uploaded = True
# If something goes wrong:
except Exception as exception:
session.abort_transaction()
file_uploaded = False
self._printer(exception)
# Done here:
return file_uploaded
async def upload_from_stream(self): pass
# ┳┓ ┓ •
# ┃┃┏┓┃┏┓╋┓┏┓┏┓
@@ -626,7 +742,7 @@ class FileObjectManagementModel:
async def make_public(self): pass
async def mak_private(self): pass
async def make_private(self): pass
# *****************************************************************************************************************
@@ -652,21 +768,40 @@ if __name__ == "__main__":
await files_mongo.connect()
my_fs = FileObjectManagementModel()
user_a = CoreUserInfoModel(
userId = 1
user_bhopli = CoreUserInfoModel(
fullName = "Bhopli Narangi",
userId = 1,
entityId = 2,
billingAccountId = 3,
departmentId = 4,
branchId = 5,
industry = "technology"
)
user_b = CoreUserInfoModel(
userId = 6
user_polki = CoreUserInfoModel(
fullName = "Polki Muchhwaali",
userId = 6,
entityId = 7,
billingAccountId = 8,
departmentId = 9,
branchId = 10,
# industry = "finance"
)
owned_files = await my_fs.list_shared_files(
file_info = await my_fs.info(
mongo_conn = files_mongo,
user_info = user_b
file_object_id = "67598d48c1bf89b25695f20b"
)
if owned_files is None: print("ERROR!")
else: print(f"FILES ({len(owned_files)}):", json.to_string(owned_files, default = str))
print("SUCCESS:", file_info.success)
print("MESSAGE:", file_info.message)
print("FILE INFO:", json.to_string(file_info.result.model_dump(), default = str))
is_owner = my_fs.has_permission(
mongo_conn = files_mongo,
user_info = user_polki,
file_object_info = file_info.result,
permission = "write"
)
print("HAS PERMISSION:", is_owner.model_dump_json(indent = 4))
asyncio.run(main())