TiagoRG
/
sio-2425
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

small conflict solved

This commit is contained in:
RubenCGomes 2024-12-17 10:02:26 +00:00
parent 34588e7568 a0173d1b01
commit 6c1de74a12
No known key found for this signature in database
20 changed files with 207 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
delivery2/client/bin/rep_create_session
View File

@ -1,10 +1,14 @@
#!/bin/python3
import base64
import os
import sys
import argparse
import logging
import json
import requests
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import hashes
from subject import main
@ -36,15 +40,15 @@ def createSession(args):
args = parser.parse_args()
if not args.org or not args.username or not args.password or not args.credentials or not args.session:
logger.error("Need organization, username, password, credentials and session file")
if not args.org or not args.username or not args.credentials or not args.session:
logger.error("Need organization, username, credentials and session file")
sys.exit(1)
if (not os.path.isfile(BASE_DIR + args.credentials)):
if not os.path.isfile(BASE_DIR + args.credentials):
logger.error("File '" + args.credentials + "' not found.")
sys.exit(1)
session = {'org' : args.org, 'username' : args.username, 'password' : args.password, 'credentials_file' : args.credentials}
session = {'org' : args.org, 'username' : args.username}
#print( type(json.dumps(session)))
@ -55,6 +59,29 @@ def createSession(args):
logger.error("Failed to obtain response from server")
sys.exit(-1)
response = req.json()
challenge = response['challenge']
with open(BASE_DIR + args.credentials, 'rb') as f:
try:
key = load_pem_private_key(f.read(), password=args.password.encode("utf-8") if args.password else None)
except ValueError:
logger.error("Invalid password")
sys.exit(-1)
signature = key.sign(
challenge.encode('utf-8'),
padding.PKCS1v15(),
hashes.SHA256()
)
try:
req = requests.post(f'http://{state['REP_ADDRESS']}/user/login', json=json.dumps({'signature' : base64.b64encode(signature).decode('utf-8')}), headers={'Authorization': response['token']})
req.raise_for_status()
except requests.exceptions.RequestException as errex:
logger.error("Failed to obtain response from server")
sys.exit(-1)
with open(BASE_DIR + args.session, 'w') as f:
json.dump(req.json(), f)

4
delivery2/client/bin/rep_decrypt_file
View File

@ -30,7 +30,7 @@ def decryptFile(args):
sys.exit(1)
# If first argument is not a file or not found
if (not os.path.isfile(BASE_DIR + args.encrypted)):
if not os.path.isfile(BASE_DIR + args.encrypted):
logger.error("File '" + args.encrypted + "' not found.")
sys.exit(1)
@ -42,7 +42,7 @@ def decryptFile(args):
print(args.metadata)
metadata = json.loads(args.metadata)
content = symmetric_encryption.decrypt_file(metadata['key'].encode(), BASE_DIR + args.encrypted, 'file.txt')
content = symmetric_encryption.decrypt_file(metadata['key'].encode(), BASE_DIR + args.encrypted)
# Send decrypted content to stdout
sys.stdout.write(content)

2
delivery2/client/bin/rep_get_doc_file
View File

@ -63,7 +63,7 @@ def getDoc(args):
#Get file with file_handle provided by metadata
try:
file = requests.get(f'http://{state['REP_ADDRESS']}/get/' + metadata['file_handle'] + '/content')
file = requests.get(f'http://{state['REP_ADDRESS']}/file/get/' + metadata['file_handle'] + '/content')
file.raise_for_status()
except requests.exceptions.RequestException as errex:
logger.error("Failed to obtain response from server.")

3
delivery2/client/bin/rep_subject_credentials
View File

@ -13,6 +13,9 @@ logger.setLevel(logging.INFO)
BASE_DIR = os.path.join(os.path.expanduser('~'), '.sio/')
if not os.path.exists(BASE_DIR):
os.makedirs(BASE_DIR)
# Generate a key pair for a subject
# password - file for public key, file for private key
def generateKeyPair(args):

2
delivery2/client/bin/reset_database
View File

@ -23,7 +23,7 @@ def reset(args):
parser.add_argument("-r", '--repo', nargs=1, help="Address:Port of the repository")
parser.add_argument("-v", '--verbose', help="Increase verbosity", action="store_true")
parser.add_argument('password', nargs='?', default=None)
parser.add_argument('password', nargs='?', default="123")
args = parser.parse_args()

2
delivery2/client/tests/test_client.py
View File

@ -40,7 +40,7 @@ def test_rep_list_orgs():
def test_rep_create_session():
# Test the rep_create_session command
process = subprocess.Popen(f"{DELIVERY_PATH}/client/bin/rep_create_session org1 username password pub.pem session.json", shell=True)
process = subprocess.Popen(f"{DELIVERY_PATH}/client/bin/rep_create_session org1 username password priv.pem session.json", shell=True)
process.wait()
assert process.returncode == 0

4
delivery2/lib/symmetric_encryption.py
View File

@ -34,7 +34,7 @@ def encrypt_file(input_file, output_file=None):
# Function to decrypt a file
def decrypt_file(key, input_file, output_file=None):
def decrypt_file(key, input_file, output_file=None) -> str:
plaintext_content = b""
with open(input_file, 'rb') as infile:
@ -62,5 +62,5 @@ def decrypt_file(key, input_file, output_file=None):
# Finalize decryption
plaintext_content += decryptor.finalize()
return plaintext_content
return plaintext_content.decode('utf-8', errors='ignore')

2
delivery2/server/app.py
View File

@ -32,7 +32,7 @@ def index():
@app.route("/reset", methods=["POST"])
def reset():
password = request.json["password"]
password = request.json.get("password")
if password != "123":
return jsonify({"error": "Invalid password"}), 403
try:

2
delivery2/server/models/file.py
View File

@ -26,9 +26,9 @@ class File(db_connection.Model):
"name": self.name,
"created_at": self.created_at,
"acl": self.acl,
"deleter_id": self.deleter_id,
"key": self.key,
"alg": self.alg,
"org": {"id": self.org.id, "name": self.org.name},
"creator": {"id": self.creator.id, "username": self.creator.username},
"deleter": {"id": self.deleter.id, "username": self.deleter.username} if self.deleter else None
}

6
delivery2/server/models/session.py
View File

@ -10,6 +10,8 @@ class Session(db_connection.Model):
roles = db_connection.Column(db_connection.JSON, default=list)
created_at = db_connection.Column(db_connection.DateTime, server_default=db_connection.func.now())
updated_at = db_connection.Column(db_connection.DateTime, server_default=db_connection.func.now(), server_onupdate=db_connection.func.now())
challenge = db_connection.Column(db_connection.String(255), unique=True)
verified = db_connection.Column(db_connection.Boolean, default=False)
def to_dict(self):
return {
@ -18,5 +20,7 @@ class Session(db_connection.Model):
"org_id": self.org_id,
"token": self.token,
"created_at": self.created_at,
"updated_at": self.updated_at
"updated_at": self.updated_at,
"challenge": self.challenge,
"verified": self.verified
}

2
delivery2/server/models/user.py
View File

@ -19,6 +19,6 @@ class User(db_connection.Model):
"username": self.username,
"full_name": self.full_name,
"email": self.email,
"role": self.role,
"roles": self.roles,
"orgs": [{"id": org_id, "name": org_data["name"], "status": org_data["status"]} for org_id, org_data in self.orgs.items()],
}

8
delivery2/server/routes/file.py
View File

@ -40,7 +40,6 @@ def file_get_metadata(document_handle: str):
@file_bp.route("/upload/metadata", methods=["POST"])
def file_upload_metadata():
session_token = request.headers.get("Authorization")
print(session_token)
session = SessionService.validate_session(session_token, required_perms=[Perm.DOC_NEW])
if isinstance(session, tuple):
return session
@ -196,10 +195,9 @@ def file_acl():
if role not in org.roles:
return jsonify({"error": "Role not found"}), 404
try:
RoleService.change_perm_on_role_in_file(file, role, perm, operation)
except ValueError as e:
return jsonify({"error": str(e)}), 400
acl = RoleService.change_perm_on_role_in_file(file, role, perm, operation)
if isinstance(acl, tuple):
return acl
return jsonify(file.to_dict()), 200

77
delivery2/server/routes/role.py
View File

@ -49,10 +49,9 @@ def role_list_users(role):
if not org:
return jsonify({"error": "Organization not found"}), 404
try:
users = RoleService.get_users_in_role(org, role)
except ValueError as e:
return jsonify({"error": str(e)}), 400
users = RoleService.get_users_in_role(org, role)
if isinstance(users, tuple):
return users
return jsonify(users), 200
@ -70,10 +69,9 @@ def role_list_perms(role):
if not org:
return jsonify({"error": "Organization not found"}), 404
try:
perms = RoleService.get_perms_for_role(org, role)
except ValueError as e:
return jsonify({"error": str(e)}), 400
perms = RoleService.get_perms_for_role(org, role, return_str=True)
if isinstance(perms, tuple):
return perms
return jsonify(perms), 200
@ -98,10 +96,9 @@ def role_suspend(role):
if not org:
return jsonify({"error": "Organization not found"}), 404
try:
RoleService.change_role_status(org, role, "suspended")
except ValueError as e:
return jsonify({"error": str(e)}), 400
status = RoleService.change_role_status(org, role, "suspended")
if isinstance(status, tuple):
return status
return jsonify({"message": "Role suspended"}), 200
@ -127,10 +124,9 @@ def role_activate(role):
if not org:
return jsonify({"error": "Organization not found"}), 404
try:
RoleService.change_role_status(org, role, "active")
except ValueError as e:
return jsonify({"error": str(e)}), 400
status = RoleService.change_role_status(org, role, "active")
if isinstance(status, tuple):
return status
return jsonify({"message": "Role activated"}), 200
@ -160,10 +156,9 @@ def role_user_add(role, username):
if not user:
return jsonify({"error": "User not found"}), 404
try:
RoleService.add_user_to_role(role, org, user)
except ValueError as e:
return jsonify({"error": str(e)}), 400
role = RoleService.add_user_to_role(role, org, user)
if isinstance(role, tuple):
return role
return jsonify({"message": "User added to role"}), 200
@ -193,10 +188,9 @@ def role_user_remove(role, username):
if not user:
return jsonify({"error": "User not found"}), 404
try:
RoleService.remove_user_from_role(role, org, user)
except ValueError as e:
return jsonify({"error": str(e)}), 400
role = RoleService.remove_user_from_role(role, org, user)
if isinstance(role, tuple):
return role
return jsonify({"message": "User removed from role"}), 200
@ -222,10 +216,9 @@ def role_perm_add(role, perm):
if not org:
return jsonify({"error": "Organization not found"}), 404
try:
RoleService.change_perm_on_role(org, role, Perm.from_str(perm), PermOperation.ADD)
except ValueError as e:
return jsonify({"error": str(e)}), 400
role = RoleService.change_perm_on_role(org, role, Perm.from_str(perm), PermOperation.ADD)
if isinstance(role, tuple):
return role
return jsonify({"message": "Permission added to role"}), 200
@ -251,10 +244,9 @@ def role_perm_remove(role, perm):
if not org:
return jsonify({"error": "Organization not found"}), 404
try:
RoleService.change_perm_on_role(org, role, Perm.from_str(perm), PermOperation.REMOVE)
except ValueError as e:
return jsonify({"error": str(e)}), 400
role = RoleService.change_perm_on_role(org, role, Perm.from_str(perm), PermOperation.REMOVE)
if isinstance(role, tuple):
return role
return jsonify({"message": "Permission removed from role"}), 200
@ -272,10 +264,9 @@ def role_session_assume(role):
if not RoleService.get_role(session.org_id, role):
return jsonify({"error": "Role not found"}), 404
try:
SessionService.change_role(session, role, "add")
except ValueError as e:
return jsonify({"error": str(e)}), 400
session = SessionService.change_role(session, role, "add")
if isinstance(session, tuple):
return session
return jsonify(session.to_dict()), 200
@ -293,10 +284,9 @@ def role_session_drop(role):
if not RoleService.get_role(session.org_id, role):
return jsonify({"error": "Role not found"}), 404
try:
SessionService.change_role(session, role, "drop")
except ValueError as e:
return jsonify({"error": str(e)}), 400
session = SessionService.change_role(session, role, "drop")
if isinstance(session, tuple):
return session
return jsonify(session.to_dict()), 200
@ -328,8 +318,7 @@ def perm_list_roles(perm):
if not org:
return jsonify({"error": "Organization not found"}), 404
try:
roles = RoleService.get_roles_for_perm(org, Perm(perm))
except ValueError as e:
return jsonify({"error": str(e)}), 400
roles = RoleService.get_roles_for_perm(org, Perm(perm))
if isinstance(roles, tuple):
return roles
return jsonify(roles), 200

42
delivery2/server/routes/user.py
View File

@ -1,4 +1,7 @@
import base64
import json
from cryptography.exceptions import InvalidSignature
from flask import Blueprint, request, jsonify
from services import UserService, SessionService, OrganizationService, RoleService
from utils import Perm
@ -11,19 +14,38 @@ def user_login():
if type(data) is str:
data = json.loads(data)
if "username" not in data or "org" not in data:
return jsonify({"error": "Missing required fields"}), 400
if "username" in data and "org" in data:
user = UserService.get_user_by_username(data["username"])
if not user:
return jsonify({"error": "User not found"}), 404
user = UserService.get_user_by_username(data["username"])
if not user:
return jsonify({"error": "User not found"}), 404
org = OrganizationService.get_organization_by_name(data["org"])
if not org:
return jsonify({"error": "Organization not found"}), 404
org = OrganizationService.get_organization_by_name(data["org"])
if not org:
return jsonify({"error": "Organization not found"}), 404
session = SessionService.create_session(user, org)
return jsonify(session.to_dict()), 201
session = SessionService.create_session(user, org)
return jsonify(session.to_dict()), 201
elif session_token := request.headers.get("Authorization"):
session = SessionService.get_session(session_token)
if not session:
return jsonify({"error": "Not authenticated"}), 401
if session.verified:
return jsonify(session.to_dict()), 200
if not "signature" in data:
return jsonify({"error": "Missing required fields"}), 400
signature = data["signature"]
signature = base64.b64decode(signature)
session = SessionService.verify_session(session_token, signature)
if isinstance(session, tuple):
return session
return jsonify(session.to_dict()), 200
return jsonify({"error": "Missing required fields"}), 400
@user_bp.route("/logout", methods=["POST"])

7
delivery2/server/services/orgs.py
View File

@ -34,7 +34,8 @@ class OrganizationService:
Perm.ROLE_UP,
Perm.ROLE_MOD
]),
"users": [user.id]
"users": [],
"status": "active"
}
}
@ -54,7 +55,7 @@ class OrganizationService:
db.refresh(organization)
UserService().add_org_to_user(user, organization)
RoleService().add_user_to_role(user, organization, "manager")
RoleService().add_user_to_role("manager", organization, user)
UserService().add_public_key_to_user(user, organization, public_key)
return organization
@ -97,7 +98,7 @@ class OrganizationService:
if OrganizationService.get_user_status(org, user.id) != "active":
return {"error": "User already suspended"}, 400
if user.roles[org.id] == "manager":
if user.id in RoleService.get_users_in_role(org, "manager"):
return {"error": "Cannot suspend manager"}, 400
org.users[str(user.id)]["status"] = "suspended"

60
delivery2/server/services/roles.py
View File

@ -1,14 +1,18 @@
from flask import jsonify
from database import db
from models import Organization, User, File
from sqlalchemy.orm.attributes import flag_modified
from services import FileService, OrganizationService
from utils import Perm, PermOperation
class RoleService:
@staticmethod
def create_role(org: Organization, role: str, perms: list[Perm]) -> dict:
def create_role(org: Organization, role: str, perms: list[Perm]) -> dict | tuple:
if role in org.roles:
raise ValueError(f"Role {role} already exists in organization {org.name}")
return jsonify({"error": f"Role {role} already exists in organization {org.name}"}), 400
roles = org.roles.copy()
roles[role] = {
@ -33,12 +37,12 @@ class RoleService:
return roles
@staticmethod
def change_role_status(org: Organization, role: str, status: str) -> Organization:
def change_role_status(org: Organization, role: str, status: str) -> Organization | tuple:
if role not in org.roles:
raise ValueError(f"Role {role} does not exist in organization {org.name}")
return jsonify({"error": f"Role {role} does not exist in organization {org.name}"}), 400
if org.roles[role]["status"] == status:
raise ValueError(f"Role {role} is already {status} in organization {org.name}")
return jsonify({"error": f"Role {role} is already {status} in organization {org.name}"}), 400
roles = org.roles.copy()
roles[role]["status"] = status
@ -55,13 +59,13 @@ class RoleService:
return org.roles[role]
@staticmethod
def check_role_permission(org: Organization, role: str, perm: Perm, doc_handle=None) -> bool:
def check_role_permission(org: Organization, role: str, perm: Perm, doc_handle=None) -> bool | tuple:
from services import FileService
if doc_handle:
file = FileService.get_file_by_document_handle(doc_handle)
if not file:
raise ValueError(f"Document {doc_handle} not found")
return jsonify({"error": "File not found"}), 404
if not Perm.check_perm(file.acl[role], perm.value):
return False
@ -69,7 +73,7 @@ class RoleService:
return True
if role not in org.roles:
raise ValueError(f"Role {role} does not exist in organization {org.name}")
return jsonify({"error": f"Role {role} does not exist in organization {org.name}"}), 400
role_perms = org.roles[role]["permissions"]
return Perm.check_perm(role_perms, perm.value)
@ -86,9 +90,9 @@ class RoleService:
return org.roles
@staticmethod
def get_users_in_role(org: Organization, role: str) -> list:
def get_users_in_role(org: Organization, role: str) -> list | tuple:
if role not in org.roles:
raise ValueError(f"Role {role} does not exist in organization {org.name}")
return jsonify({"error": f"Role {role} does not exist in organization {org.name}"}), 400
return org.roles[role]["users"]
@staticmethod
@ -97,7 +101,11 @@ class RoleService:
@staticmethod
def get_perms_for_role(org: Organization, role: str, return_str=False) -> list[Perm | str]:
return Perm.get_perms(org.roles[role]["permissions"], return_str)
perms_list = Perm.get_perms(org.roles[role]["permissions"], return_str)
for f in FileService.list_files_in_org(org):
perms_list.append({f.document_handle: Perm.get_perms(f.acl[role], return_str)})
return perms_list
@staticmethod
def get_roles_for_perm(org: Organization, perm: Perm) -> list:
@ -108,15 +116,15 @@ class RoleService:
return roles
@staticmethod
def add_user_to_role(role: str, org: Organization, user: User) -> User:
def add_user_to_role(role: str, org: Organization, user: User) -> User | tuple:
if role not in org.roles:
raise ValueError(f"Role {role} does not exist in organization {org.name}")
return jsonify({"error": f"Role {role} does not exist in organization {org.name}"}), 400
if user.id in org.roles[role]["users"]:
raise ValueError(f"User {user.username} already has role {role} in organization {org.name}")
return jsonify({"error": f"User {user.username} already has role {role} in organization {org.name}"}), 400
if org.roles[role]["status"] != "active":
raise ValueError(f"Role {role} is not active in organization {org.name}")
return jsonify({"error": f"Role {role} is not active in organization {org.name}"}), 400
roles = user.roles.copy()
roles[org.id] = role
@ -134,15 +142,15 @@ class RoleService:
return user
@staticmethod
def remove_user_from_role(role: str, org: Organization, user: User) -> User:
def remove_user_from_role(role: str, org: Organization, user: User) -> User | tuple:
if role not in org.roles:
raise ValueError(f"Role {role} does not exist in organization {org.name}")
return jsonify({"error": f"Role {role} does not exist in organization {org.name}"}), 400
if user.id not in org.roles[role]["users"]:
raise ValueError(f"User {user.username} does not have role {role} in organization {org.name}")
return jsonify({"error": f"User {user.username} does not have role {role} in organization {org.name}"}), 400
if org.roles[role]["status"] != "active":
raise ValueError(f"Role {role} is not active in organization {org.name}")
return jsonify({"error": f"Role {role} is not active in organization {org.name}"}), 400
roles = user.roles.copy()
roles.pop(org.id)
@ -160,15 +168,15 @@ class RoleService:
return user
@staticmethod
def change_perm_on_role(org: Organization, role: str, perm: Perm, operation: PermOperation) -> dict:
def change_perm_on_role(org: Organization, role: str, perm: Perm, operation: PermOperation) -> dict | tuple:
if Perm.get_int([perm]) <= 0b111:
raise ValueError(f"Permission {perm} is not allowed for organization's roles")
return jsonify({"error": f"Permission {perm} is not allowed for organization's roles"}), 400
if role not in org.roles:
raise ValueError(f"Role {role} does not exist in organization {org.name}")
return jsonify({"error": f"Role {role} does not exist in organization {org.name}"}), 400
if org.roles[role]["status"] != "active":
raise ValueError(f"Role {role} is not active in organization {org.name}")
return jsonify({"error": f"Role {role} is not active in organization {org.name}"}), 400
roles = org.roles.copy()
roles[role]["permissions"] = PermOperation.calc(roles[role]["permissions"], perm, operation)
@ -179,15 +187,15 @@ class RoleService:
return org.roles[role]
@staticmethod
def change_perm_on_role_in_file(file: File, role: str, perm: Perm, operation: PermOperation) -> dict:
def change_perm_on_role_in_file(file: File, role: str, perm: Perm, operation: PermOperation) -> dict | tuple:
if Perm.get_int([perm]) > 0b111:
raise ValueError(f"Permission {perm} is not allowed for files' roles")
return jsonify({"error": f"Permission {perm} is not allowed for files' roles"}), 400
if role not in file.acl:
file.acl[role] = 0
if file.acl[role] & perm.value != 0:
raise ValueError(f"Role {role} already has permission {perm} in file {file.document_handle}")
return jsonify({"error": f"Role {role} already has permission {perm} in file {file.document_handle}"}), 400
file.acl[role] = PermOperation.calc(file.acl[role], perm, operation)
flag_modified(file, "acl")

47
delivery2/server/services/sessions.py
View File

@ -1,5 +1,8 @@
import secrets
from cryptography.hazmat.primitives.serialization import load_pem_public_key
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import hashes
from sqlalchemy.orm.attributes import flag_modified
from database import db
@ -16,13 +19,35 @@ class SessionService:
user_id=user.id,
org_id=org.id,
token=secrets.token_hex(128),
roles=[]
roles=[],
challenge=secrets.token_hex(128),
verified=False
)
db.add(session)
db.commit()
db.refresh(session)
return session
@staticmethod
def verify_session(token: str, signature: bytes):
session = SessionService.get_session(token)
if not session:
return jsonify({"error": "Session not found"}), 401
public_key_pem = User.query.get(session.user_id).public_keys.get(str(session.org_id))
if not public_key_pem:
return jsonify({"error": "Public key not found"}), 404
public_key = load_pem_public_key(public_key_pem.encode())
public_key.verify(
signature,
session.challenge.encode(),
padding.PKCS1v15(),
hashes.SHA256()
)
session.challenge = None
session.verified = True
db.commit()
db.refresh(session)
@staticmethod
def get_session(token: str) -> Session | None:
return db.query(Session).filter(Session.token == token).first()
@ -43,6 +68,9 @@ class SessionService:
if not session:
return jsonify({"error": "Not authenticated"}), 401
if not session.verified:
return jsonify({"error": "Session has not yet been verified"}), 403
org = OrganizationService.get_organization(session.org_id)
if not org:
return jsonify({"error": "Organization not found"}), 404
@ -64,37 +92,38 @@ class SessionService:
org = OrganizationService.get_organization(session.org_id)
if not org:
raise ValueError(f"Organization {session.org_id} not found")
return jsonify({"error": "Organization not found"}), 404
user = User.query.get(session.user_id)
if not user:
raise ValueError(f"User {session.user_id} not found")
return jsonify({"error": "User not found"}), 404
if role not in org.roles:
raise ValueError(f"Role {role} does not exist in organization {org.name}")
return jsonify({"error": f"Role {role} does not exist in organization {org.name}"}), 404
if operation == "add":
if role not in user.roles[org.id]:
raise ValueError(f"User {user.username} does not have role {role}")
return jsonify({"error": f"User {user.username} does not have role {role}"}), 400
if role in session.roles:
raise ValueError(f"User {user.username} already has role {role} in current session")
return jsonify({"error": f"User {user.username} already has role {role} in current session"}), 400
session.roles.append(role)
elif operation == "drop":
if role not in user.roles[org.id]:
raise ValueError(f"User {user.username} does not have role {role}")
return jsonify({"error": f"User {user.username} does not have role {role}"}), 400
if role not in session.roles:
raise ValueError(f"User {user.username} does not have role {role} in current session")
return jsonify({"error": f"User {user.username} does not have role {role} in current session"}), 400
session.roles.remove(role)
else:
raise ValueError(f"Invalid operation {operation}")
return jsonify({"error": "Invalid operation"}), 400
flag_modified(session, "roles")
db.commit()
db.refresh(session)
return session
@staticmethod
def list_roles(session: Session) -> list:

3
delivery2/server/services/users.py
View File

@ -2,6 +2,7 @@ from sqlalchemy.orm.attributes import flag_modified
from database import db
from models import User, Organization
from utils import encode_public_key
class UserService:
@ -13,7 +14,7 @@ class UserService:
full_name=full_name,
email=email,
roles={},
public_keys={org.id: public_key} if org else {},
public_keys={org.id: encode_public_key(public_key)} if org else {},
orgs={org.id: {
"name": org.name,
"status": "active"

2
delivery2/server/utils/__init__.py
View File

@ -1,3 +1,3 @@
from .checks import check_valid_time
from .hashing import get_hash, get_hex_from_temp_file
from .hashing import get_hash, get_hex_from_temp_file, encode_public_key
from .perms import Perm, PermOperation

11
delivery2/server/utils/hashing.py
View File

@ -1,5 +1,6 @@
from tempfile import SpooledTemporaryFile
import cryptography.hazmat.primitives.hashes
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat, load_pem_public_key
def get_hash(data):
@ -14,3 +15,13 @@ def get_hex_from_temp_file(temp_file: SpooledTemporaryFile) -> bytes:
temp_file.seek(0)
file_data = temp_file.read()
return file_data
def encode_public_key(public_key):
if isinstance(public_key, str):
public_key = load_pem_public_key(public_key.encode('utf-8'))
return public_key.public_bytes(
encoding=cryptography.hazmat.primitives.serialization.Encoding.PEM,
format=cryptography.hazmat.primitives.serialization.PublicFormat.SubjectPublicKeyInfo
).decode('utf-8')