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

Merge pull request #102 from learnhouse/swve/eng-88-securityroles-issues-more-bugs

Browse source 
Revamp Backend Authorization + Bugfixes
This commit is contained in:
Badr B 2023-07-22 15:37:49 +02:00 • committed by GitHub
commit 9b9f46f693
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
23 changed files with 731 additions and 448 deletions
Download patch file Download diff file Expand all files Collapse all files

40
app.py
View file

@ -8,6 +8,9 @@ from fastapi.staticfiles import StaticFiles
from fastapi_jwt_auth.exceptions import AuthJWTException
from fastapi.middleware.gzip import GZipMiddleware
from src.security.rbac.rbac import authorization_verify_based_on_roles, authorization_verify_if_element_is_public, authorization_verify_if_user_is_author
from src.services.users.schemas.users import UserRolesInOrganization
# from src.services.mocks.initial import create_initial_data
@ -66,3 +69,40 @@ app.include_router(v1_router)
@app.get("/")
async def root():
return {"Message": "Welcome to LearnHouse ✨"}
@app.get("/test")
async def rootd(request: Request):
res = await authorization_verify_based_on_roles(
request=request,
user_id="user_c441e47e-5c04-4b03-9886-b0f5cb333c06",
action="read",
roles_list=[
UserRolesInOrganization(
org_id="org_e7085838-2efc-48f3-b414-77318572d9f5", role_id="role_admin"
),
],
element_id="collection_1c277b46-5a4b-440a-ac29-94b874ef7cf4",
)
return res
@app.get("/test2")
async def rootds(request: Request):
res = await authorization_verify_if_user_is_author(
request=request,
user_id="user_c441e47e-5c04-4b03-9886-b0f5cb333c06",
action="read",
element_id="course_1c277b46-5a4b-440a-ac29-94b874ef7cf4",
)
return res
@app.get("/test3")
async def rootdsc(request: Request):
res = await authorization_verify_if_element_is_public(
request=request,
user_id="anonymous",
action="read",
element_id="course_1c277b46-5a4b-440a-ac29-94b874ef7cf4",
)
return res

12
front/app/install/steps/get_started.tsx
View file

@ -9,19 +9,21 @@ function GetStarted() {
const { data: install, error: error, isLoading } = useSWR(`${getAPIUrl()}install/latest`, swrFetcher);
const router = useRouter()
function startInstallation() {
fetch(`${getAPIUrl()}install/start`, {
async function startInstallation() {
let res = await fetch(`${getAPIUrl()}install/start`, {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({})
}).then(res => res.json()).then(res => {
if (res.success) {
})
if (res.status == 200) {
mutate(`${getAPIUrl()}install/latest`)
router.refresh();
router.push(`/install?step=1`)
}
})
}

1
front/app/orgs/[orgslug]/(withmenu)/collections/new/page.tsx
View file

@ -40,6 +40,7 @@ function NewCollection(params: any) {
name: name,
description: description,
courses: selectedCourses,
public: true,
org_id: org.org_id,
};
await createCollection(collection);

2
front/app/orgs/[orgslug]/(withmenu)/collections/page.tsx
View file

@ -45,7 +45,7 @@ const CollectionsPage = async (params: any) => {
<GeneralWrapperStyled>
<div className="flex justify-between" >
<TypeOfContentTitle title="Collections" type="col" />
<AuthenticatedClientElement checkMethod='authentication'>
<AuthenticatedClientElement checkMethod='roles' orgId={org_id}>
<Link className="flex justify-center" href={getUriWithOrg(orgslug, "/collections/new")}>
<button className="rounded-md bg-black antialiased ring-offset-purple-800 p-2 px-5 my-auto font text-sm font-bold text-white drop-shadow-lg">Add Collection + </button>
</Link>

3
front/app/orgs/[orgslug]/(withmenu)/courses/courses.tsx
View file

@ -20,6 +20,7 @@ import { getCourseThumbnailMediaDirectory } from '@services/media/media';
interface CourseProps {
orgslug: string;
courses: any;
org_id: string;
}
// function to remove "course_" from the course_id
@ -49,7 +50,7 @@ function Courses(props: CourseProps) {
<GeneralWrapperStyled>
<div className='flex flex-wrap justify-between'>
<TypeOfContentTitle title="Courses" type="cou" />
<AuthenticatedClientElement checkMethod='authentication'>
<AuthenticatedClientElement checkMethod='roles' orgId={props.org_id}>
<Modal
isDialogOpen={newCourseModal}
onOpenChange={setNewCourseModal}

3
front/app/orgs/[orgslug]/(withmenu)/courses/page.tsx
View file

@ -25,13 +25,14 @@ export async function generateMetadata(
const CoursesPage = async (params: any) => {
const orgslug = params.params.orgslug;
const org = await getOrganizationContextInfo(orgslug, { revalidate: 1800, tags: ['organizations'] });
const cookieStore = cookies();
const access_token_cookie: any = cookieStore.get('access_token_cookie');
const courses = await getOrgCoursesWithAuthHeader(orgslug, { revalidate: 0, tags: ['courses'] }, access_token_cookie ? access_token_cookie.value : null);
return (
<div>
<Courses orgslug={orgslug} courses={courses} />
<Courses org_id={org.org_id} orgslug={orgslug} courses={courses} />
</div>
);
};

4
front/components/Security/AuthenticatedClientElement.tsx
View file

@ -14,7 +14,7 @@ export const AuthenticatedClientElement = (props: AuthenticatedClientElementProp
// Available roles
const org_roles_values = ["admin", "owner"];
const user_roles_values = ["role_admin", "role_super_admin"];
const user_roles_values = ["role_admin"];
@ -26,7 +26,7 @@ export const AuthenticatedClientElement = (props: AuthenticatedClientElementProp
const user_role = user_roles.find((role: any) => role.org_id == org_id);
if (org_role && user_role) {
if (org_roles_values.includes(org_role.org_role) && user_roles_values.includes(user_role.role_id)) {
if (org_roles_values.includes(org_role.org_role) || user_roles_values.includes(user_role.role_id)) {
return true;
}
else {

2
front/services/install/install.ts
View file

@ -33,7 +33,7 @@ export async function createDefaultElements() {
export async function isInstallModeEnabled() {
const result = await fetch(`${getAPIUrl()}install/latest`, RequestBody("GET", null, null));
if (result.status === 200) {
if (result.status === 200 || result.status === 404) {
return true;
}
else {

2
src/security/auth.py
View file

@ -18,7 +18,7 @@ class Settings(BaseModel):
authjwt_secret_key: str = "secret" if isDevModeEnabled() else SECRET_KEY
authjwt_token_location = {"cookies", "headers"}
authjwt_cookie_csrf_protect = False
authjwt_access_token_expires = False if isDevModeEnabled() else 3600
authjwt_access_token_expires = False if isDevModeEnabled() else 28800
authjwt_cookie_samesite = "lax"
authjwt_cookie_secure = True
authjwt_cookie_domain = get_learnhouse_config().hosting_config.cookie_config.domain

149
src/security/rbac/rbac.py Normal file
View file

@ -0,0 +1,149 @@
from typing import Literal
from fastapi import HTTPException, status, Request
from src.security.rbac.utils import check_element_type, get_id_identifier_of_element
from src.services.roles.schemas.roles import RoleInDB
from src.services.users.schemas.users import UserRolesInOrganization
async def authorization_verify_if_element_is_public(
request,
element_id: str,
user_id: str,
action: Literal["read"],
):
element_nature = await check_element_type(element_id)
# Verifies if the element is public
if (
element_nature == ("courses" or "collections")
and action == "read"
and user_id == "anonymous"
):
if element_nature == "courses":
courses = request.app.db["courses"]
course = await courses.find_one({"course_id": element_id})
if course["public"]:
return True
else:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User rights (public content) : You don't have the right to perform this action",
)
if element_nature == "collections":
collections = request.app.db["collections"]
collection = await collections.find_one({"collection_id": element_id})
if collection["public"]:
return True
else:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User rights (public content) : You don't have the right to perform this action",
)
else:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User rights (public content) : You don't have the right to perform this action",
)
async def authorization_verify_if_user_is_author(
request,
user_id: str,
action: Literal["read", "update", "delete", "create"],
element_id: str,
):
if action == "update" or "delete" or "read":
element_nature = await check_element_type(element_id)
elements = request.app.db[element_nature]
element_identifier = await get_id_identifier_of_element(element_id)
element = await elements.find_one({element_identifier: element_id})
if user_id in element["authors"]:
return True
else:
return False
else:
return False
async def authorization_verify_based_on_roles(
request: Request,
user_id: str,
action: Literal["read", "update", "delete", "create"],
roles_list: list[UserRolesInOrganization],
element_id: str,
):
element_type = await check_element_type(element_id)
element = request.app.db[element_type]
roles = request.app.db["roles"]
# Get the element
element_identifier = await get_id_identifier_of_element(element_id)
element = await element.find_one({element_identifier: element_id})
# Get the roles of the user
roles_id_list = [role["role_id"] for role in roles_list]
roles = await roles.find({"role_id": {"$in": roles_id_list}}).to_list(length=100)
async def checkRoles():
# Check Roles
for role in roles:
role = RoleInDB(**role)
if role.elements[element_type][f"action_{action}"] is True:
return True
else:
return False
async def checkOrgRoles():
# Check Org Roles
users = request.app.db["users"]
user = await users.find_one({"user_id": user_id})
if element is not None:
for org in user["orgs"]:
if org["org_id"] == element["org_id"]:
if org["org_role"] == "owner" or org["org_role"] == "editor":
return True
else:
return False
if await checkRoles() or await checkOrgRoles():
return True
else:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User rights (roless) : You don't have the right to perform this action",
)
async def authorization_verify_based_on_roles_and_authorship(
request: Request,
user_id: str,
action: Literal["read", "update", "delete", "create"],
roles_list: list[UserRolesInOrganization],
element_id: str,
):
isAuthor = await authorization_verify_if_user_is_author(
request, user_id, action, element_id
)
isRole = await authorization_verify_based_on_roles(
request, user_id, action, roles_list, element_id
)
if isAuthor or isRole:
return True
else:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User rights (roles & authorship) : You don't have the right to perform this action",
)
async def authorization_verify_if_user_is_anon(user_id: str):
if user_id == "anonymous":
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You should be logged in to perform this action",
)

45
src/security/rbac/utils.py Normal file
View file

@ -0,0 +1,45 @@
from fastapi import HTTPException, status
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
"""
if element_id.startswith("course_"):
return "courses"
elif element_id.startswith("user_"):
return "users"
elif element_id.startswith("house_"):
return "houses"
elif element_id.startswith("org_"):
return "organizations"
elif element_id.startswith("coursechapter_"):
return "coursechapters"
elif element_id.startswith("collection_"):
return "collections"
elif element_id.startswith("activity_"):
return "activities"
else:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="User rights : Issue verifying element nature",
)
async def get_singular_form_of_element(element_id):
element_type = await check_element_type(element_id)
if element_type == "activities":
return "activity"
else:
singular_form_element = element_type[:-1]
return singular_form_element
async def get_id_identifier_of_element(element_id):
singular_form_element = await get_singular_form_of_element(element_id)
if singular_form_element == "ogranizations":
return "org_id"
else:
return str(singular_form_element) + "_id"

121
src/security/security.py
View file

@ -1,10 +1,7 @@
from fastapi import HTTPException, status, Request
from passlib.context import CryptContext
from passlib.hash import pbkdf2_sha256
from config.config import get_learnhouse_config
from src.services.roles.schemas.roles import RoleInDB
from src.services.users.schemas.users import UserInDB, UserRolesInOrganization
### 🔒 JWT ##############################################################
@ -30,122 +27,4 @@ async def security_verify_password(plain_password: str, hashed_password: str):
### 🔒 Passwords Hashing ##############################################################
### 🔒 Roles checking ##############################################################
async def verify_user_rights_with_roles(
request: Request, action: str, user_id: str, element_id: str, element_org_id: str
):
"""
Check if the user has the right to perform the action on the element
"""
request.app.db["roles"]
users = request.app.db["users"]
user = await users.find_one({"user_id": user_id})
#########
# Users existence verification
#########
if not user and user_id != "anonymous":
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="User rights : User not found"
)
# Check if user is anonymous
if user_id == "anonymous":
return False
# Get User
user: UserInDB = UserInDB(**await users.find_one({"user_id": user_id}))
#########
# Organization Roles verification
#########
for org in user.orgs:
if org.org_id == element_org_id:
# Check if user is owner or reader of the organization
if org.org_role == ("owner" or "editor"):
return True
#########
# Roles verification
#########
user_roles = user.roles
if action != "create":
return await check_user_role_org_with_element_org(
request, element_id, user_roles, action
)
# If no role is found, raise an error
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User rights : You don't have the right to perform this action",
)
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
"""
if element_id.startswith("course_"):
return "courses"
elif element_id.startswith("user_"):
return "users"
elif element_id.startswith("house_"):
return "houses"
elif element_id.startswith("org_"):
return "organizations"
elif element_id.startswith("coursechapter_"):
return "coursechapters"
elif element_id.startswith("collection_"):
return "collections"
elif element_id.startswith("activity_"):
return "activities"
else:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="User rights : Issue verifying element nature",
)
async def check_user_role_org_with_element_org(
request: Request,
element_id: str,
roles_list: list[UserRolesInOrganization],
action: str,
):
element_type = await check_element_type(element_id)
element = request.app.db[element_type]
roles = request.app.db["roles"]
# get singular element type
singular_form_element = element_type[:-1]
element_type_id = singular_form_element + "_id"
element_org = await element.find_one({element_type_id: element_id})
for role in roles_list:
# Check if The role belongs to the same organization as the element
role_db = await roles.find_one({"role_id": role.role_id})
role = RoleInDB(**role_db)
if (role.org_id == element_org["org_id"]) or role.org_id == "*":
# Check if user has the right role
for role in roles_list:
role_db = await roles.find_one({"role_id": role.role_id})
role = RoleInDB(**role_db)
if role.elements[element_type][f"action_{action}"]:
return True
else:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User rights (roles) : You don't have the right to perform this action",
)
### 🔒 Roles checking ##############################################################

113
src/services/courses/activities/activities.py
View file

@ -1,6 +1,11 @@
from typing import Literal
from pydantic import BaseModel
from src.security.security import verify_user_rights_with_roles
from src.services.users.schemas.users import PublicUser
from src.security.rbac.rbac import (
authorization_verify_based_on_roles,
authorization_verify_if_element_is_public,
authorization_verify_if_user_is_anon,
)
from src.services.users.schemas.users import AnonymousUser, PublicUser
from fastapi import HTTPException, status, Request
from uuid import uuid4
from datetime import datetime
@ -40,23 +45,26 @@ async def create_activity(
):
activities = request.app.db["activities"]
courses = request.app.db["courses"]
users = request.app.db["users"]
# get user
user = await users.find_one({"user_id": current_user.user_id})
# generate activity_id
activity_id = str(f"activity_{uuid4()}")
hasRoleRights = await verify_user_rights_with_roles(
request, "create", current_user.user_id, activity_id, org_id
# verify activity rights
await authorization_verify_based_on_roles(
request,
current_user.user_id,
"create",
user["roles"],
activity_id,
)
# get course_id from activity
course = await courses.find_one({"chapters": coursechapter_id})
if not hasRoleRights:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Roles : Insufficient rights to perform this action",
)
# create activity
activity = ActivityInDB(
**activity_object.dict(),
@ -86,29 +94,10 @@ async def get_activity(request: Request, activity_id: str, current_user: PublicU
# get course_id from activity
coursechapter_id = activity["coursechapter_id"]
course = await courses.find_one({"chapters": coursechapter_id})
isCoursePublic = course["public"]
isAuthor = current_user.user_id in course["authors"]
if isAuthor:
activity = ActivityInDB(**activity)
return activity
await courses.find_one({"chapters": coursechapter_id})
# verify course rights
hasRoleRights = await verify_user_rights_with_roles(
request,
"read",
current_user.user_id,
activity_id,
element_org_id=activity["org_id"],
)
if not hasRoleRights and not isCoursePublic:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Roles : Insufficient rights to perform this action",
)
await verify_rights(request, activity["course_id"], current_user, "read")
if not activity:
raise HTTPException(
@ -128,14 +117,9 @@ async def update_activity(
activities = request.app.db["activities"]
activity = await activities.find_one({"activity_id": activity_id})
# verify course rights
await verify_user_rights_with_roles(
request,
"update",
current_user.user_id,
activity_id,
element_org_id=activity["org_id"],
)
await verify_rights(request, activity_id, current_user, "update")
if activity:
creationDate = activity["creationDate"]
@ -171,13 +155,7 @@ async def delete_activity(request: Request, activity_id: str, current_user: Publ
activity = await activities.find_one({"activity_id": activity_id})
# verify course rights
await verify_user_rights_with_roles(
request,
"delete",
current_user.user_id,
activity_id,
element_org_id=activity["org_id"],
)
await verify_rights(request, activity_id, current_user, "delete")
if not activity:
raise HTTPException(
@ -217,3 +195,48 @@ async def get_activities(
]
return activities
#### Security ####################################################
async def verify_rights(
request: Request,
activity_id: str, # course_id in case of read
current_user: PublicUser | AnonymousUser,
action: Literal["create", "read", "update", "delete"],
):
if action == "read":
if current_user.user_id == "anonymous":
await authorization_verify_if_element_is_public(
request, activity_id, current_user.user_id, action
)
else:
users = request.app.db["users"]
user = await users.find_one({"user_id": current_user.user_id})
await authorization_verify_if_user_is_anon(current_user.user_id)
await authorization_verify_based_on_roles(
request,
current_user.user_id,
action,
user["roles"],
activity_id,
)
else:
users = request.app.db["users"]
user = await users.find_one({"user_id": current_user.user_id})
await authorization_verify_if_user_is_anon(current_user.user_id)
await authorization_verify_based_on_roles(
request,
current_user.user_id,
action,
user["roles"],
activity_id,
)
#### Security ####################################################

20
src/services/courses/activities/pdf.py
View file

@ -1,4 +1,4 @@
from src.security.security import verify_user_rights_with_roles
from src.security.rbac.rbac import authorization_verify_based_on_roles
from src.services.courses.activities.uploads.pdfs import upload_pdf
from src.services.users.users import PublicUser
from src.services.courses.activities.activities import ActivityInDB
@ -16,6 +16,10 @@ async def create_documentpdf_activity(
):
activities = request.app.db["activities"]
courses = request.app.db["courses"]
users = request.app.db["users"]
# get user
user = await users.find_one({"user_id": current_user.user_id})
# generate activity_id
activity_id = str(f"activity_{uuid4()}")
@ -64,14 +68,12 @@ async def create_documentpdf_activity(
updateDate=str(datetime.now()),
)
hasRoleRights = await verify_user_rights_with_roles(
request, "create", current_user.user_id, activity_id, element_org_id=org_id
)
if not hasRoleRights:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Roles : Insufficient rights to perform this action",
await authorization_verify_based_on_roles(
request,
current_user.user_id,
"create",
user["roles"],
activity_id,
)
# create activity

40
src/services/courses/activities/video.py
View file

@ -1,7 +1,9 @@
from typing import Literal
from pydantic import BaseModel
from src.security.security import verify_user_rights_with_roles
from src.security.rbac.rbac import (
authorization_verify_based_on_roles,
)
from src.services.courses.activities.uploads.videos import upload_video
from src.services.users.users import PublicUser
from src.services.courses.activities.activities import ActivityInDB
@ -19,6 +21,10 @@ async def create_video_activity(
):
activities = request.app.db["activities"]
courses = request.app.db["courses"]
users = request.app.db["users"]
# get user
user = await users.find_one({"user_id": current_user.user_id})
# generate activity_id
activity_id = str(f"activity_{uuid4()}")
@ -75,14 +81,12 @@ async def create_video_activity(
updateDate=str(datetime.now()),
)
hasRoleRights = await verify_user_rights_with_roles(
request, "create", current_user.user_id, activity_id, element_org_id=org_id
)
if not hasRoleRights:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Roles : Insufficient rights to perform this action",
await authorization_verify_based_on_roles(
request,
current_user.user_id,
"create",
user["roles"],
activity_id,
)
# create activity
@ -122,6 +126,10 @@ async def create_external_video_activity(
):
activities = request.app.db["activities"]
courses = request.app.db["courses"]
users = request.app.db["users"]
# get user
user = await users.find_one({"user_id": current_user.user_id})
# generate activity_id
activity_id = str(f"activity_{uuid4()}")
@ -157,14 +165,12 @@ async def create_external_video_activity(
updateDate=str(datetime.now()),
)
hasRoleRights = await verify_user_rights_with_roles(
request, "create", current_user.user_id, activity_id, element_org_id=org_id
)
if not hasRoleRights:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Roles : Insufficient rights to perform this action",
await authorization_verify_based_on_roles(
request,
current_user.user_id,
"create",
user["roles"],
activity_id,
)
# create activity

223
src/services/courses/chapters.py
View file

@ -1,11 +1,16 @@
from datetime import datetime
from typing import List
from typing import List, Literal
from uuid import uuid4
from pydantic import BaseModel
from src.security.auth import non_public_endpoint
from src.security.rbac.rbac import (
authorization_verify_based_on_roles,
authorization_verify_based_on_roles_and_authorship,
authorization_verify_if_element_is_public,
authorization_verify_if_user_is_anon,
)
from src.services.courses.courses import Course
from src.services.courses.activities.activities import ActivityInDB
from src.security.security import verify_user_rights_with_roles
from src.services.users.users import PublicUser
from fastapi import HTTPException, status, Request
@ -29,6 +34,7 @@ class CourseChapterMetaData(BaseModel):
chapters: dict
activities: object
#### Classes ####################################################
####################################################
@ -36,35 +42,60 @@ class CourseChapterMetaData(BaseModel):
####################################################
async def create_coursechapter(request: Request, coursechapter_object: CourseChapter, course_id: str, current_user: PublicUser):
async def create_coursechapter(
request: Request,
coursechapter_object: CourseChapter,
course_id: str,
current_user: PublicUser,
):
courses = request.app.db["courses"]
print(course_id)
users = request.app.db["users"]
# get course org_id and verify rights
course = await courses.find_one({"course_id": course_id})
await courses.find_one({"course_id": course_id})
user = await users.find_one({"user_id": current_user.user_id})
# generate coursechapter_id with uuid4
coursechapter_id = str(f"coursechapter_{uuid4()}")
hasRoleRights = await verify_user_rights_with_roles(request, "create", current_user.user_id, coursechapter_id, course["org_id"])
hasRoleRights = await authorization_verify_based_on_roles(
request, current_user.user_id, "create", user["roles"], course_id
)
if not hasRoleRights:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="Roles : Insufficient rights to perform this action")
status_code=status.HTTP_409_CONFLICT,
detail="Roles : Insufficient rights to perform this action",
)
coursechapter = CourseChapterInDB(coursechapter_id=coursechapter_id, creationDate=str(
datetime.now()), updateDate=str(datetime.now()), course_id=course_id, **coursechapter_object.dict())
coursechapter = CourseChapterInDB(
coursechapter_id=coursechapter_id,
creationDate=str(datetime.now()),
updateDate=str(datetime.now()),
course_id=course_id,
**coursechapter_object.dict(),
)
courses.update_one({"course_id": course_id}, {
"$addToSet": {"chapters": coursechapter_id, "chapters_content": coursechapter.dict()}})
courses.update_one(
{"course_id": course_id},
{
"$addToSet": {
"chapters": coursechapter_id,
"chapters_content": coursechapter.dict(),
}
},
)
return coursechapter.dict()
async def get_coursechapter(request: Request, coursechapter_id: str, current_user: PublicUser):
async def get_coursechapter(
request: Request, coursechapter_id: str, current_user: PublicUser
):
courses = request.app.db["courses"]
coursechapter = await courses.find_one(
{"chapters_content.coursechapter_id": coursechapter_id})
{"chapters_content.coursechapter_id": coursechapter_id}
)
if coursechapter:
# verify course rights
@ -75,64 +106,87 @@ async def get_coursechapter(request: Request, coursechapter_id: str, current_use
else:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="CourseChapter does not exist")
status_code=status.HTTP_409_CONFLICT, detail="CourseChapter does not exist"
)
async def update_coursechapter(request: Request, coursechapter_object: CourseChapter, coursechapter_id: str, current_user: PublicUser):
async def update_coursechapter(
request: Request,
coursechapter_object: CourseChapter,
coursechapter_id: str,
current_user: PublicUser,
):
courses = request.app.db["courses"]
coursechapter = await courses.find_one(
{"chapters_content.coursechapter_id": coursechapter_id})
{"chapters_content.coursechapter_id": coursechapter_id}
)
if coursechapter:
# verify course rights
await verify_rights(request, coursechapter["course_id"], current_user, "update")
coursechapter = CourseChapterInDB(coursechapter_id=coursechapter_id, creationDate=str(
datetime.now()), updateDate=str(datetime.now()), course_id=coursechapter["course_id"], **coursechapter_object.dict())
coursechapter = CourseChapterInDB(
coursechapter_id=coursechapter_id,
creationDate=str(datetime.now()),
updateDate=str(datetime.now()),
course_id=coursechapter["course_id"],
**coursechapter_object.dict(),
)
courses.update_one({"chapters_content.coursechapter_id": coursechapter_id}, {
"$set": {"chapters_content.$": coursechapter.dict()}})
courses.update_one(
{"chapters_content.coursechapter_id": coursechapter_id},
{"$set": {"chapters_content.$": coursechapter.dict()}},
)
return coursechapter
else:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="Coursechapter does not exist")
status_code=status.HTTP_409_CONFLICT, detail="Coursechapter does not exist"
)
async def delete_coursechapter(request: Request, coursechapter_id: str, current_user: PublicUser):
async def delete_coursechapter(
request: Request, coursechapter_id: str, current_user: PublicUser
):
courses = request.app.db["courses"]
course = await courses.find_one(
{"chapters_content.coursechapter_id": coursechapter_id})
{"chapters_content.coursechapter_id": coursechapter_id}
)
if course:
# verify course rights
await verify_rights(request, course["course_id"], current_user, "delete")
# Remove coursechapter from course
await courses.update_one({"course_id": course["course_id"]}, {
"$pull": {"chapters": coursechapter_id}})
await courses.update_one({"chapters_content.coursechapter_id": coursechapter_id}, {
"$pull": {"chapters_content": {"coursechapter_id": coursechapter_id}}})
await courses.update_one(
{"course_id": course["course_id"]},
{"$pull": {"chapters": coursechapter_id}},
)
await courses.update_one(
{"chapters_content.coursechapter_id": coursechapter_id},
{"$pull": {"chapters_content": {"coursechapter_id": coursechapter_id}}},
)
return {"message": "Coursechapter deleted"}
else:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="Course does not exist")
status_code=status.HTTP_409_CONFLICT, detail="Course does not exist"
)
####################################################
# Misc
####################################################
async def get_coursechapters(request: Request, course_id: str, page: int = 1, limit: int = 10):
async def get_coursechapters(
request: Request, course_id: str, page: int = 1, limit: int = 10
):
courses = request.app.db["courses"]
course = await courses.find_one({"course_id": course_id})
@ -144,19 +198,26 @@ async def get_coursechapters(request: Request, course_id: str, page: int = 1, li
return coursechapters
async def get_coursechapters_meta(request: Request, course_id: str, current_user: PublicUser):
async def get_coursechapters_meta(
request: Request, course_id: str, current_user: PublicUser
):
courses = request.app.db["courses"]
activities = request.app.db["activities"]
await non_public_endpoint(current_user)
coursechapters = await courses.find_one({"course_id": course_id}, {"chapters": 1, "chapters_content": 1, "_id": 0})
await verify_rights(request, course_id, current_user, "read")
coursechapters = await courses.find_one(
{"course_id": course_id}, {"chapters": 1, "chapters_content": 1, "_id": 0}
)
coursechapters = coursechapters
if not coursechapters:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="Course does not exist")
status_code=status.HTTP_409_CONFLICT, detail="Course does not exist"
)
# activities
coursechapter_activityIds_global = []
@ -165,7 +226,6 @@ async def get_coursechapters_meta(request: Request, course_id: str, current_user
chapters = {}
if coursechapters["chapters_content"]:
for coursechapter in coursechapters["chapters_content"]:
coursechapter = CourseChapterInDB(**coursechapter)
coursechapter_activityIds = []
@ -174,37 +234,55 @@ async def get_coursechapters_meta(request: Request, course_id: str, current_user
coursechapter_activityIds_global.append(activity)
chapters[coursechapter.coursechapter_id] = {
"id": coursechapter.coursechapter_id, "name": coursechapter.name, "activityIds": coursechapter_activityIds
"id": coursechapter.coursechapter_id,
"name": coursechapter.name,
"activityIds": coursechapter_activityIds,
}
# activities
activities_list = {}
for activity in await activities.find({"activity_id": {"$in": coursechapter_activityIds_global}}).to_list(length=100):
for activity in await activities.find(
{"activity_id": {"$in": coursechapter_activityIds_global}}
).to_list(length=100):
activity = ActivityInDB(**activity)
activities_list[activity.activity_id] = {
"id": activity.activity_id, "name": activity.name, "type": activity.type, "content": activity.content
"id": activity.activity_id,
"name": activity.name,
"type": activity.type,
"content": activity.content,
}
final = {
"chapters": chapters,
"chapterOrder": coursechapters["chapters"],
"activities": activities_list
"activities": activities_list,
}
return final
async def update_coursechapters_meta(request: Request, course_id: str, coursechapters_metadata: CourseChapterMetaData, current_user: PublicUser):
async def update_coursechapters_meta(
request: Request,
course_id: str,
coursechapters_metadata: CourseChapterMetaData,
current_user: PublicUser,
):
courses = request.app.db["courses"]
await verify_rights(request, course_id, current_user, "update")
# update chapters in course
await courses.update_one({"course_id": course_id}, {
"$set": {"chapters": coursechapters_metadata.chapterOrder}})
await courses.update_one(
{"course_id": course_id},
{"$set": {"chapters": coursechapters_metadata.chapterOrder}},
)
if coursechapters_metadata.chapters is not None:
for coursechapter_id, chapter_metadata in coursechapters_metadata.chapters.items():
filter_query = {
"chapters_content.coursechapter_id": coursechapter_id}
for (
coursechapter_id,
chapter_metadata,
) in coursechapters_metadata.chapters.items():
filter_query = {"chapters_content.coursechapter_id": coursechapter_id}
update_query = {
"$set": {
"chapters_content.$.activities": chapter_metadata["activityIds"]
@ -213,30 +291,61 @@ async def update_coursechapters_meta(request: Request, course_id: str, coursecha
result = await courses.update_one(filter_query, update_query)
if result.matched_count == 0:
# handle error when no documents are matched by the filter query
print(
f"No documents found for course chapter ID {coursechapter_id}")
print(f"No documents found for course chapter ID {coursechapter_id}")
return {"detail": "coursechapters metadata updated"}
#### Security ####################################################
async def verify_rights(request: Request, course_id: str, current_user: PublicUser, action: str):
async def verify_rights(
request: Request,
course_id: str,
current_user: PublicUser,
action: Literal["read", "update", "delete"],
):
courses = request.app.db["courses"]
users = request.app.db["users"]
user = await users.find_one({"user_id": current_user.user_id})
course = await courses.find_one({"course_id": course_id})
if not course:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="Course does not exist")
status_code=status.HTTP_409_CONFLICT, detail="Course does not exist"
)
hasRoleRights = await verify_user_rights_with_roles(request, action, current_user.user_id, course_id, course["org_id"])
isAuthor = current_user.user_id in course["authors"]
if action == "read":
if current_user.user_id == "anonymous":
await authorization_verify_if_element_is_public(
request, course_id, current_user.user_id, action
)
else:
users = request.app.db["users"]
user = await users.find_one({"user_id": current_user.user_id})
if not hasRoleRights and not isAuthor:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, detail="Roles/Ownership : Insufficient rights to perform this action")
await authorization_verify_if_user_is_anon(current_user.user_id)
await authorization_verify_based_on_roles_and_authorship(
request,
current_user.user_id,
action,
user["roles"],
course_id,
)
else:
users = request.app.db["users"]
user = await users.find_one({"user_id": current_user.user_id})
await authorization_verify_if_user_is_anon(current_user.user_id)
await authorization_verify_based_on_roles_and_authorship(
request,
current_user.user_id,
action,
user["roles"],
course_id,
)
return True
#### Security ####################################################

36
src/services/courses/collections.py
View file

@ -1,8 +1,8 @@
from typing import List
from typing import List, Literal
from uuid import uuid4
from pydantic import BaseModel
from src.security.rbac.rbac import authorization_verify_based_on_roles_and_authorship, authorization_verify_if_user_is_anon
from src.services.users.users import PublicUser
from src.security.security import verify_user_rights_with_roles
from fastapi import HTTPException, status, Request
#### Classes ####################################################
@ -12,11 +12,13 @@ class Collection(BaseModel):
name: str
description: str
courses: List[str] # course_id
public: bool
org_id: str # org_id
class CollectionInDB(Collection):
collection_id: str
authors: List[str] # user_id
#### Classes ####################################################
@ -81,7 +83,11 @@ async def create_collection(
# generate collection_id with uuid4
collection_id = str(f"collection_{uuid4()}")
collection = CollectionInDB(collection_id=collection_id, **collection_object.dict())
collection = CollectionInDB(
collection_id=collection_id,
authors=[current_user.user_id],
**collection_object.dict(),
)
collection_in_db = await collections.insert_one(collection.dict())
@ -169,6 +175,11 @@ async def get_collections(
print(org_id)
if current_user.user_id == "anonymous":
all_collections = collections.find(
{"org_id": org_id, "public": True}, {"_id": 0}
)
else:
# get all collections from database without ObjectId
all_collections = (
collections.find({"org_id": org_id})
@ -177,8 +188,6 @@ async def get_collections(
.limit(limit)
)
await verify_collection_rights(request, "*", current_user, "read", org_id)
# create list of collections and include courses in each collection
collections_list = []
for collection in await all_collections.to_list(length=100):
@ -207,11 +216,12 @@ async def verify_collection_rights(
request: Request,
collection_id: str,
current_user: PublicUser,
action: str,
action: Literal["create", "read", "update", "delete"],
org_id: str,
):
collections = request.app.db["collections"]
users = request.app.db["users"]
user = await users.find_one({"user_id": current_user.user_id})
collection = await collections.find_one({"collection_id": collection_id})
if not collection and action != "create" and collection_id != "*":
@ -223,17 +233,11 @@ async def verify_collection_rights(
if current_user.user_id == "anonymous" and action == "read":
return True
hasRoleRights = await verify_user_rights_with_roles(
request, action, current_user.user_id, collection_id, org_id
)
await authorization_verify_if_user_is_anon(current_user.user_id)
if not hasRoleRights:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You do not have rights to this Collection",
await authorization_verify_based_on_roles_and_authorship(
request, current_user.user_id, action, user["roles"], collection_id
)
return True
#### Security ####################################################

95
src/services/courses/courses.py
View file

@ -1,12 +1,17 @@
import json
from typing import List, Optional
from typing import List, Literal, Optional
from uuid import uuid4
from pydantic import BaseModel
from src.security.rbac.rbac import (
authorization_verify_based_on_roles,
authorization_verify_based_on_roles_and_authorship,
authorization_verify_if_element_is_public,
authorization_verify_if_user_is_anon,
)
from src.services.courses.activities.activities import ActivityInDB
from src.services.courses.thumbnails import upload_thumbnail
from src.services.users.schemas.users import AnonymousUser
from src.services.users.users import PublicUser
from src.security.security import verify_user_rights_with_roles
from fastapi import HTTPException, Request, status, UploadFile
from datetime import datetime
@ -144,7 +149,6 @@ async def get_course_meta(request: Request, course_id: str, current_user: Public
trail = await trails.find_one(
{"courses.course_id": course_id, "user_id": current_user.user_id}
)
print(trail)
if trail:
# get only the course where course_id == course_id
trail_course = next(
@ -169,6 +173,8 @@ async def create_course(
thumbnail_file: UploadFile | None = None,
):
courses = request.app.db["courses"]
users = request.app.db["users"]
user = await users.find_one({"user_id": current_user.user_id})
# generate course_id with uuid4
course_id = str(f"course_{uuid4()}")
@ -176,10 +182,16 @@ async def create_course(
# TODO(fix) : the implementation here is clearly not the best one (this entire function)
course_object.org_id = org_id
course_object.chapters_content = []
await verify_user_rights_with_roles(
request, "create", current_user.user_id, course_id, org_id
await authorization_verify_based_on_roles(
request,
current_user.user_id,
"create",
user["roles"],
course_id,
)
if thumbnail_file and thumbnail_file.filename:
name_in_disk = (
f"{course_id}_thumbnail_{uuid4()}.{thumbnail_file.filename.split('.')[-1]}"
@ -214,12 +226,13 @@ async def update_course_thumbnail(
current_user: PublicUser,
thumbnail_file: UploadFile | None = None,
):
# verify course rights
await verify_rights(request, course_id, current_user, "update")
courses = request.app.db["courses"]
course = await courses.find_one({"course_id": course_id})
# verify course rights
await verify_rights(request, course_id, current_user, "update")
# TODO(fix) : the implementation here is clearly not the best one
if course:
creationDate = course["creationDate"]
@ -254,13 +267,13 @@ async def update_course_thumbnail(
async def update_course(
request: Request, course_object: Course, course_id: str, current_user: PublicUser
):
# verify course rights
await verify_rights(request, course_id, current_user, "update")
courses = request.app.db["courses"]
course = await courses.find_one({"course_id": course_id})
# verify course rights
await verify_rights(request, course_id, current_user, "update")
if course:
creationDate = course["creationDate"]
authors = course["authors"]
@ -289,13 +302,13 @@ async def update_course(
async def delete_course(request: Request, course_id: str, current_user: PublicUser):
# verify course rights
await verify_rights(request, course_id, current_user, "delete")
courses = request.app.db["courses"]
course = await courses.find_one({"course_id": course_id})
# verify course rights
await verify_rights(request, course_id, current_user, "delete")
if not course:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="Course does not exist"
@ -364,41 +377,37 @@ async def verify_rights(
request: Request,
course_id: str,
current_user: PublicUser | AnonymousUser,
action: str,
action: Literal["create", "read", "update", "delete"],
):
courses = request.app.db["courses"]
course = await courses.find_one({"course_id": course_id})
isAuthor = current_user.user_id in course["authors"]
if isAuthor:
return True
if (
current_user.user_id == "anonymous"
and course["public"] is True
and action == "read"
):
return True
if not course:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Course/CourseChapter does not exist",
if action == "read":
if current_user.user_id == "anonymous":
await authorization_verify_if_element_is_public(
request, course_id, current_user.user_id, action
)
else:
users = request.app.db["users"]
user = await users.find_one({"user_id": current_user.user_id})
hasRoleRights = await verify_user_rights_with_roles(
request, action, current_user.user_id, course_id, course["org_id"]
await authorization_verify_based_on_roles_and_authorship(
request,
current_user.user_id,
action,
user["roles"],
course_id,
)
else:
users = request.app.db["users"]
user = await users.find_one({"user_id": current_user.user_id})
if not hasRoleRights and not isAuthor:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Roles/Ownership : Insufficient rights to perform this action",
await authorization_verify_if_user_is_anon(current_user.user_id)
await authorization_verify_based_on_roles_and_authorship(
request,
current_user.user_id,
action,
user["roles"],
course_id,
)
return True
#### Security ####################################################

71
src/services/install/install.py
View file

@ -125,8 +125,8 @@ async def install_default_elements(request: Request, data: dict):
roles = request.app.db["roles"]
# check if default roles ADMIN_ROLE and USER_ROLE already exist
admin_role = await roles.find_one({"role_id": "role_super_admin"})
user_role = await roles.find_one({"role_id": "role_user"})
admin_role = await roles.find_one({"role_id": "role_admin"})
user_role = await roles.find_one({"role_id": "role_member"})
if admin_role is not None or user_role is not None:
raise HTTPException(
@ -135,61 +135,8 @@ async def install_default_elements(request: Request, data: dict):
)
# get default roles
SUPER_ADMIN_ROLE = RoleInDB(
name="SuperAdmin Role",
description="This role grants all permissions to the user",
elements=Elements(
courses=Permission(
action_create=True,
action_read=True,
action_update=True,
action_delete=True,
),
users=Permission(
action_create=True,
action_read=True,
action_update=True,
action_delete=True,
),
houses=Permission(
action_create=True,
action_read=True,
action_update=True,
action_delete=True,
),
collections=Permission(
action_create=True,
action_read=True,
action_update=True,
action_delete=True,
),
organizations=Permission(
action_create=True,
action_read=True,
action_update=True,
action_delete=True,
),
coursechapters=Permission(
action_create=True,
action_read=True,
action_update=True,
action_delete=True,
),
activities=Permission(
action_create=True,
action_read=True,
action_update=True,
action_delete=True,
),
),
org_id="*",
role_id="role_super_admin",
created_at=str(datetime.now()),
updated_at=str(datetime.now()),
)
ADMIN_ROLE = RoleInDB(
name="SuperAdmin Role",
name="Admin Role",
description="This role grants all permissions to the user",
elements=Elements(
courses=Permission(
@ -236,13 +183,13 @@ async def install_default_elements(request: Request, data: dict):
),
),
org_id="*",
role_id="role_super_admin",
role_id="role_admin",
created_at=str(datetime.now()),
updated_at=str(datetime.now()),
)
USER_ROLE = RoleInDB(
name="User role",
name="Member Role",
description="This role grants read-only permissions to the user",
elements=Elements(
courses=Permission(
@ -289,16 +236,14 @@ async def install_default_elements(request: Request, data: dict):
),
),
org_id="*",
role_id="role_user",
role_id="role_member",
created_at=str(datetime.now()),
updated_at=str(datetime.now()),
)
try:
# insert default roles
await roles.insert_many(
[ADMIN_ROLE.dict(), USER_ROLE.dict(), SUPER_ADMIN_ROLE.dict()]
)
await roles.insert_many([USER_ROLE.dict(), ADMIN_ROLE.dict()])
return True
except Exception:
@ -386,7 +331,7 @@ async def install_create_organization_user(
orgs = [UserOrganization(org_id=org_id, org_role="owner")]
# Give role
roles = [UserRolesInOrganization(role_id="role_super_admin", org_id=org_id)]
roles = [UserRolesInOrganization(role_id="role_admin", org_id=org_id)]
# Create the user
user = UserInDB(

23
src/services/orgs/orgs.py
View file

@ -1,5 +1,10 @@
import json
from typing import Literal
from uuid import uuid4
from src.security.rbac.rbac import (
authorization_verify_based_on_roles,
authorization_verify_if_user_is_anon,
)
from src.services.orgs.logos import upload_org_logo
from src.services.orgs.schemas.orgs import (
Organization,
@ -8,7 +13,6 @@ from src.services.orgs.schemas.orgs import (
)
from src.services.users.schemas.users import UserOrganization
from src.services.users.users import PublicUser
from src.security.security import verify_user_rights_with_roles
from fastapi import HTTPException, UploadFile, status, Request
@ -197,9 +201,12 @@ async def verify_org_rights(
request: Request,
org_id: str,
current_user: PublicUser,
action: str,
action: Literal["create", "read", "update", "delete"],
):
orgs = request.app.db["organizations"]
users = request.app.db["users"]
user = await users.find_one({"user_id": current_user.user_id})
org = await orgs.find_one({"org_id": org_id})
@ -208,17 +215,11 @@ async def verify_org_rights(
status_code=status.HTTP_409_CONFLICT, detail="Organization does not exist"
)
hasRoleRights = await verify_user_rights_with_roles(
request, action, current_user.user_id, org_id, org_id
)
await authorization_verify_if_user_is_anon(current_user.user_id)
if not hasRoleRights:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You do not have rights to this organization",
await authorization_verify_based_on_roles(
request, current_user.user_id, action, user["roles"], org_id
)
return True
#### Security ####################################################

3
src/services/roles/roles.py
View file

@ -1,5 +1,6 @@
from typing import Literal
from uuid import uuid4
from src.security.rbac.rbac import authorization_verify_if_user_is_anon
from src.services.roles.schemas.roles import Role, RoleInDB
from src.services.users.schemas.users import PublicUser
from fastapi import HTTPException, status, Request
@ -85,6 +86,8 @@ async def verify_user_permissions_on_roles(
status_code=status.HTTP_401_UNAUTHORIZED, detail="Roles : Not authenticated"
)
await authorization_verify_if_user_is_anon(current_user.user_id)
if action == "create":
if "owner" in [org.org_role for org in current_user.orgs]:
return True

6
src/services/users/schemas/users.py
View file

@ -41,6 +41,9 @@ class UserInDB(User):
creation_date: str
update_date: str
def __getitem__(self, item):
return getattr(self, item)
@ -54,6 +57,9 @@ class PublicUser(User):
class AnonymousUser(BaseModel):
user_id: str = "anonymous"
username: str = "anonymous"
roles: list[UserRolesInOrganization] = [
UserRolesInOrganization(org_id="anonymous", role_id="role_anonymous")
]

131
src/services/users/users.py
View file

@ -2,24 +2,39 @@ from datetime import datetime
from typing import Literal
from uuid import uuid4
from fastapi import HTTPException, Request, status
from src.security.rbac.rbac import authorization_verify_based_on_roles, authorization_verify_if_user_is_anon
from src.security.security import security_hash_password, security_verify_password
from src.services.users.schemas.users import PasswordChangeForm, PublicUser, User, UserOrganization, UserRolesInOrganization, UserWithPassword, UserInDB
from src.services.users.schemas.users import (
PasswordChangeForm,
PublicUser,
User,
UserOrganization,
UserRolesInOrganization,
UserWithPassword,
UserInDB,
)
async def create_user(request: Request, current_user: PublicUser | None, user_object: UserWithPassword, org_slug: str):
async def create_user(
request: Request,
current_user: PublicUser | None,
user_object: UserWithPassword,
org_slug: str,
):
users = request.app.db["users"]
isUsernameAvailable = await users.find_one({"username": user_object.username})
isEmailAvailable = await users.find_one({"email": user_object.email})
if isUsernameAvailable:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="Username already exists")
status_code=status.HTTP_409_CONFLICT, detail="Username already exists"
)
if isEmailAvailable:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="Email already exists")
status_code=status.HTTP_409_CONFLICT, detail="Email already exists"
)
# Generate user_id with uuid4
user_id = str(f"user_{uuid4()}")
@ -42,11 +57,12 @@ async def create_user(request: Request, current_user: PublicUser | None, user_o
# If the org does not exist, raise an error
if not isOrgExists:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="You are trying to create a user in an organization that does not exist")
status_code=status.HTTP_409_CONFLICT,
detail="You are trying to create a user in an organization that does not exist",
)
org_id = isOrgExists["org_id"]
# Create initial orgs list with the org_id passed in
orgs = [UserOrganization(org_id=org_id, org_role="member")]
@ -54,8 +70,14 @@ async def create_user(request: Request, current_user: PublicUser | None, user_o
roles = [UserRolesInOrganization(role_id="role_member", org_id=org_id)]
# Create the user
user = UserInDB(user_id=user_id, creation_date=str(datetime.now()),
update_date=str(datetime.now()), orgs=orgs, roles=roles, **user_object.dict())
user = UserInDB(
user_id=user_id,
creation_date=str(datetime.now()),
update_date=str(datetime.now()),
orgs=orgs,
roles=roles,
**user_object.dict(),
)
# Insert the user into the database
await users.insert_one(user.dict())
@ -75,12 +97,15 @@ async def read_user(request: Request, current_user: PublicUser, user_id: str):
# If the user does not exist, raise an error
if not isUserExists:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="User does not exist")
status_code=status.HTTP_409_CONFLICT, detail="User does not exist"
)
return User(**isUserExists)
async def update_user(request: Request, user_id: str, user_object: User,current_user: PublicUser):
async def update_user(
request: Request, user_id: str, user_object: User, current_user: PublicUser
):
users = request.app.db["users"]
# Verify rights
@ -92,7 +117,8 @@ async def update_user(request: Request, user_id: str, user_object: User,current
if not isUserExists:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="User does not exist")
status_code=status.HTTP_409_CONFLICT, detail="User does not exist"
)
# okay if username is not changed
if isUserExists["username"] == user_object.username:
@ -101,11 +127,13 @@ async def update_user(request: Request, user_id: str, user_object: User,current
else:
if isUsernameAvailable:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="Username already used")
status_code=status.HTTP_409_CONFLICT, detail="Username already used"
)
if isEmailAvailable:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="Email already used")
status_code=status.HTTP_409_CONFLICT, detail="Email already used"
)
updated_user = {"$set": user_object.dict()}
users.update_one({"user_id": user_id}, updated_user)
@ -113,8 +141,12 @@ async def update_user(request: Request, user_id: str, user_object: User,current
return User(**user_object.dict())
async def update_user_password(request: Request, current_user: PublicUser, user_id: str, password_change_form: PasswordChangeForm):
async def update_user_password(
request: Request,
current_user: PublicUser,
user_id: str,
password_change_form: PasswordChangeForm,
):
users = request.app.db["users"]
isUserExists = await users.find_one({"user_id": user_id})
@ -124,11 +156,15 @@ async def update_user_password(request: Request, current_user: PublicUser, user_
if not isUserExists:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="User does not exist")
status_code=status.HTTP_409_CONFLICT, detail="User does not exist"
)
if not await security_verify_password(password_change_form.old_password, isUserExists["password"]):
if not await security_verify_password(
password_change_form.old_password, isUserExists["password"]
):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail="Wrong password")
status_code=status.HTTP_401_UNAUTHORIZED, detail="Wrong password"
)
new_password = await security_hash_password(password_change_form.new_password)
@ -148,7 +184,8 @@ async def delete_user(request: Request, current_user: PublicUser, user_id: str):
if not isUserExists:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="User does not exist")
status_code=status.HTTP_409_CONFLICT, detail="User does not exist"
)
await users.delete_one({"user_id": user_id})
@ -157,18 +194,21 @@ async def delete_user(request: Request, current_user: PublicUser, user_id: str):
# Utils & Security functions
async def security_get_user(request: Request, email: str):
users = request.app.db["users"]
user = await users.find_one({"email": email})
if not user:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="User with Email does not exist")
status_code=status.HTTP_409_CONFLICT,
detail="User with Email does not exist",
)
return UserInDB(**user)
async def get_userid_by_username(request: Request, username: str):
users = request.app.db["users"]
@ -176,10 +216,12 @@ async def get_userid_by_username(request: Request, username: str):
if not user:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="User does not exist")
status_code=status.HTTP_409_CONFLICT, detail="User does not exist"
)
return user["user_id"]
async def get_user_by_userid(request: Request, user_id: str):
users = request.app.db["users"]
@ -187,32 +229,36 @@ async def get_user_by_userid(request: Request, user_id: str):
if not user:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="User does not exist")
status_code=status.HTTP_409_CONFLICT, detail="User does not exist"
)
user = User(**user)
return user
async def get_profile_metadata(request: Request, user):
users = request.app.db["users"]
request.app.db["roles"]
user = await users.find_one({"user_id": user['user_id']})
user = await users.find_one({"user_id": user["user_id"]})
if not user:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="User does not exist")
status_code=status.HTTP_409_CONFLICT, detail="User does not exist"
)
return {
"user_object": PublicUser(**user),
"roles": "random"
}
return {"user_object": PublicUser(**user), "roles": "random"}
# Verification of the user's permissions on the roles
async def verify_user_rights_on_user(request: Request, current_user: PublicUser, action: Literal["create", "read", "update", "delete"], user_id: str):
async def verify_user_rights_on_user(
request: Request,
current_user: PublicUser,
action: Literal["create", "read", "update", "delete"],
user_id: str,
):
users = request.app.db["users"]
user = UserInDB(**await users.find_one({"user_id": user_id}))
@ -220,6 +266,9 @@ async def verify_user_rights_on_user(request: Request, current_user: PublicUser,
return True
if action == "read":
await authorization_verify_if_user_is_anon(current_user.user_id)
if current_user.user_id == user_id:
return True
@ -230,27 +279,35 @@ async def verify_user_rights_on_user(request: Request, current_user: PublicUser,
return False
if action == "update":
await authorization_verify_if_user_is_anon(current_user.user_id)
if current_user.user_id == user_id:
return True
for org in current_user.orgs:
if org.org_id in [org.org_id for org in user.orgs]:
if org.org_role == "owner":
return True
# TODO: Verify user roles on the org
await authorization_verify_based_on_roles(
request, current_user.user_id, "update", user["roles"], user_id
)
return False
if action == "delete":
await authorization_verify_if_user_is_anon(current_user.user_id)
if current_user.user_id == user_id:
return True
for org in current_user.orgs:
if org.org_id in [org.org_id for org in user.orgs]:
if org.org_role == "owner":
return True
# TODO: Verify user roles on the org
await authorization_verify_based_on_roles(
request, current_user.user_id, "update", user["roles"], user_id
)