rzmk/learnhouse
1
0
Fork 0
mirror of https://github.com/rzmk/learnhouse.git synced 2025-12-19 04:19:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: collections and courses remaining bugs

Browse source
This commit is contained in:
swve 2024-02-04 21:02:07 +01:00
parent 4245e61df8
commit 212c50768f
8 changed files with 155 additions and 113 deletions
Download patch file Download diff file Expand all files Collapse all files

22
apps/api/src/security/rbac/rbac.py
View file

@ -19,9 +19,8 @@ async def authorization_verify_if_element_is_public(
):
element_nature = await check_element_type(element_uuid)
# Verifies if the element is public
if element_nature == ("courses" or "collections") and action == "read":
if element_nature == ("courses") and action == "read":
if element_nature == "courses":
print("looking for course")
statement = select(Course).where(
Course.public == True, Course.course_uuid == element_uuid
)
@ -29,20 +28,29 @@ async def authorization_verify_if_element_is_public(
if course:
return True
else:
return False
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User rights : You don't have the right to perform this action",
)
if element_nature == "collections" and action == "read":
if element_nature == "collections":
statement = select(Collection).where(
Collection.public == True, Collection.collection_uuid == element_uuid
)
collection = db_session.exec(statement).first()
if collection:
return True
else:
return False
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User rights : You don't have the right to perform this action",
)
else:
return False
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User rights : You don't have the right to perform this action",
)
# Tested and working

1
apps/api/src/security/rbac/utils.py
View file

@ -5,7 +5,6 @@ async def check_element_type(element_id):
"""
Check if the element is a course, a user, a house or a collection, by checking its prefix
"""
print("element_id", element_id)
if element_id.startswith("course_"):
return "courses"
elif element_id.startswith("user_"):

1
apps/api/src/services/courses/activities/activities.py
View file

@ -223,7 +223,6 @@ async def rbac_check(
res = await authorization_verify_if_element_is_public(
request, course_uuid, action, db_session
)
print('res',res)
return res
else:
res = await authorization_verify_based_on_roles_and_authorship(

61
apps/api/src/services/courses/collections.py
View file

@ -26,7 +26,10 @@ from fastapi import HTTPException, status, Request
async def get_collection(
request: Request, collection_uuid: str, current_user: PublicUser, db_session: Session
request: Request,
collection_uuid: str,
current_user: PublicUser,
db_session: Session,
) -> CollectionRead:
statement = select(Collection).where(Collection.collection_uuid == collection_uuid)
collection = db_session.exec(statement).first()
@ -42,11 +45,23 @@ async def get_collection(
)
# get courses in collection
statement = (
statement_all = (
select(Course)
.join(CollectionCourse, Course.id == CollectionCourse.course_id)
.distinct(Course.id)
)
statement_public = (
select(Course)
.join(CollectionCourse, Course.id == CollectionCourse.course_id)
.where(CollectionCourse.org_id == collection.org_id, Course.public == True)
)
if current_user.id == 0:
statement = statement_public
else:
statement = statement_all
courses = db_session.exec(statement).all()
collection = CollectionRead(**collection.dict(), courses=courses)
@ -180,7 +195,10 @@ async def update_collection(
async def delete_collection(
request: Request, collection_uuid: str, current_user: PublicUser, db_session: Session
request: Request,
collection_uuid: str,
current_user: PublicUser,
db_session: Session,
):
statement = select(Collection).where(Collection.collection_uuid == collection_uuid)
collection = db_session.exec(statement).first()
@ -216,23 +234,40 @@ async def get_collections(
page: int = 1,
limit: int = 10,
) -> List[CollectionRead]:
# RBAC check
await rbac_check(request, "collection_x", current_user, "read", db_session)
statement = (
statement_public = select(Collection).where(
Collection.org_id == org_id, Collection.public == True
)
statement_all = (
select(Collection).where(Collection.org_id == org_id).distinct(Collection.id)
)
if current_user.id == 0:
statement = statement_public
else:
statement = statement_all
collections = db_session.exec(statement).all()
collections_with_courses = []
for collection in collections:
statement = (
statement_all = (
select(Course)
.join(CollectionCourse, Course.id == CollectionCourse.course_id)
.distinct(Course.id)
)
statement_public = (
select(Course)
.join(CollectionCourse, Course.id == CollectionCourse.course_id)
.where(CollectionCourse.org_id == org_id, Course.public == True)
)
if current_user.id == 0:
statement = statement_public
else:
# RBAC check
statement = statement_all
courses = db_session.exec(statement).all()
collection = CollectionRead(**collection.dict(), courses=courses)
@ -256,8 +291,11 @@ async def rbac_check(
res = await authorization_verify_if_element_is_public(
request, collection_uuid, action, db_session
)
print('res',res)
return res
if res == False:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User rights : You are not allowed to read this collection",
)
else:
res = await authorization_verify_based_on_roles_and_authorship(
request, current_user.id, action, collection_uuid, db_session
@ -276,4 +314,3 @@ async def rbac_check(
## 🔒 RBAC Utils ##