SelfPrivacy/selfprivacy-rest-api
1
0
Fork 0
mirror of https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git synced 2025-02-06 16:10:39 +00:00
Code Issues Projects Releases Wiki Activity

fix: from review and make new error messages

Browse source
This commit is contained in:
dettlaff 2024-12-06 03:05:28 +04:00
parent 9de8822bd7
commit bd070a3c1d
8 changed files with 183 additions and 131 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
selfprivacy_api/actions/ssh.py
View file

@ -8,13 +8,6 @@ from selfprivacy_api.repositories.users.exceptions import UserNotFound
from selfprivacy_api.utils import ensure_ssh_and_users_fields_exist
def enable_ssh():
with WriteUserData() as data:
if "ssh" not in data:
data["ssh"] = {}
data["ssh"]["enable"] = True
class UserdataSshSettings(BaseModel):
"""Settings for the SSH."""
@ -23,6 +16,37 @@ class UserdataSshSettings(BaseModel):
rootKeys: list[str] = []
class KeyNotFound(Exception):
"""Key not found"""
@staticmethod
def get_error_message() -> str:
return "Key not found"
class KeyAlreadyExists(Exception):
"""Key already exists"""
@staticmethod
def get_error_message() -> str:
return "Key already exists"
class InvalidPublicKey(Exception):
"""Invalid public key"""
@staticmethod
def get_error_message() -> str:
return "Invalid key type. Only ssh-ed25519, ssh-rsa and ecdsa are supported"
def enable_ssh():
with WriteUserData() as data:
if "ssh" not in data:
data["ssh"] = {}
data["ssh"]["enable"] = True
def get_ssh_settings() -> UserdataSshSettings:
with ReadUserData() as data:
if "ssh" not in data:
@ -48,18 +72,6 @@ def set_ssh_settings(
data["ssh"]["passwordAuthentication"] = password_authentication
class KeyAlreadyExists(Exception):
"""Key already exists"""
pass
class InvalidPublicKey(Exception):
"""Invalid public key"""
pass
def create_ssh_key(username: str, ssh_key: str):
"""Create a new ssh key"""
@ -96,12 +108,6 @@ def create_ssh_key(username: str, ssh_key: str):
raise UserNotFound()
class KeyNotFound(Exception):
"""Key not found"""
pass
def remove_ssh_key(username: str, ssh_key: str):
"""Delete a ssh key"""

19
selfprivacy_api/actions/users.py
View file

@ -1,6 +1,7 @@
"""Actions to manage the users."""
import re
import uuid
from typing import Optional
from selfprivacy_api.models.user import UserDataUser, UserDataUserOrigin
@ -8,6 +9,7 @@ from selfprivacy_api.models.user import UserDataUser, UserDataUserOrigin
from selfprivacy_api.utils import is_username_forbidden
from selfprivacy_api.actions.ssh import get_ssh_keys
from selfprivacy_api.repositories.users.json_user_repository import JsonUserRepository
from selfprivacy_api.repositories.users import ACTIVE_USERS_PROVIDER
from selfprivacy_api.repositories.users.exceptions import (
@ -18,6 +20,17 @@ from selfprivacy_api.repositories.users.exceptions import (
)
class ApiUsingWrongUserRepository(Exception):
"""
API is using a too old or unfinished user repository. Are you debugging?
"""
@staticmethod
def get_error_message() -> str:
"""Return text message error."""
return "API is using a too old or unfinished user repository"
def get_users(
exclude_primary: bool = False,
exclude_root: bool = False,
@ -66,7 +79,9 @@ def create_user(
raise UsernameTooLong("Username must be less than 32 characters")
if ACTIVE_USERS_PROVIDER != JsonUserRepository: # for ssh management
JsonUserRepository.create_user(username=username, password="legacy")
JsonUserRepository.create_user(
username=username, password=uuid.uuid4()
) # random password for legacy
return ACTIVE_USERS_PROVIDER.create_user(
username=username,
@ -125,6 +140,6 @@ def get_user_by_username(username: str) -> Optional[UserDataUser]:
def generate_password_reset_link(username: str) -> str:
if ACTIVE_USERS_PROVIDER == JsonUserRepository:
return "Error: API using old user manager provider!"
raise ApiUsingWrongUserRepository
return ACTIVE_USERS_PROVIDER.generate_password_reset_link(username=username)

1
selfprivacy_api/graphql/common_types/user.py
View file

@ -30,7 +30,6 @@ class User:
email: Optional[str] = None
directmemberof: Optional[list[str]] = None
memberof: Optional[list[str]] = None
# userHomeFolderspace: UserHomeFolderUsage
@strawberry.type

164
selfprivacy_api/graphql/mutations/users_mutations.py
View file

@ -39,6 +39,22 @@ from selfprivacy_api.repositories.users.exceptions import (
from selfprivacy_api import PLEASE_UPDATE_APP_TEXT
FAILED_TO_SETUP_PASSWORD_TEXT = "Failed to set a password for a user. The problem occurred due to an old version of the SelfPrivacy app."
def return_failed_mutation_return(
error: str,
code: int = 400,
username: str = None,
) -> UserMutationReturn:
return UserMutationReturn(
success=False,
message=str(error),
code=code,
user=get_user_by_username(username) if username else None,
)
@strawberry.input
class UserMutationInput:
"""Input type for user mutation"""
@ -74,47 +90,38 @@ class UsersMutations:
directmemberof=user.directmemberof,
memberof=user.memberof,
)
except PasswordIsEmpty as e:
return UserMutationReturn(
success=False,
message=str(e),
code=400,
except (
PasswordIsEmpty,
UsernameNotAlphanumeric,
UsernameTooLong,
InvalidConfiguration,
) as error:
return return_failed_mutation_return(
error=error,
message=error.get_description(),
)
except UsernameForbidden as e:
return UserMutationReturn(
success=False,
message=str(e),
except UsernameForbidden as error:
return return_failed_mutation_return(
message=error.get_error_message(),
code=409,
username=user.username,
)
except UserAlreadyExists as error:
return return_failed_mutation_return(
message=error.get_error_message(),
code=409,
)
except UsernameNotAlphanumeric as e:
return UserMutationReturn(
success=False,
message=str(e),
code=400,
)
except UsernameTooLong as e:
return UserMutationReturn(
success=False,
message=str(e),
code=400,
)
except InvalidConfiguration as e:
return UserMutationReturn(
success=False,
message=str(e),
code=400,
)
except UserAlreadyExists as e:
return UserMutationReturn(
success=False,
message=str(e),
code=409,
user=get_user_by_username(user.username),
if user.password:
return return_failed_mutation_return(
message=f"{FAILED_TO_SETUP_PASSWORD_TEXT} {PLEASE_UPDATE_APP_TEXT}",
code=201,
username=user.username,
)
return UserMutationReturn(
success=True,
message=PLEASE_UPDATE_APP_TEXT if user.password else "User created",
message="User created",
code=201,
user=get_user_by_username(user.username),
)
@ -123,18 +130,13 @@ class UsersMutations:
def delete_user(self, username: str) -> GenericMutationReturn:
try:
delete_user_action(username)
except UserNotFound as e:
return GenericMutationReturn(
success=False,
message=str(e),
except UserNotFound as error:
return return_failed_mutation_return(
message=error.get_error_message(),
code=404,
)
except UserIsProtected as e:
return GenericMutationReturn(
success=False,
message=str(e),
code=400,
)
except UserIsProtected as error:
return return_failed_mutation_return(error=error)
return GenericMutationReturn(
success=True,
@ -154,24 +156,15 @@ class UsersMutations:
directmemberof=user.directmemberof,
memberof=user.memberof,
)
except PasswordIsEmpty as e:
return UserMutationReturn(
success=False,
message=str(e),
code=400,
except (PasswordIsEmpty, SelfPrivacyAppIsOutdate) as error:
return return_failed_mutation_return(
message=error.get_error_message(),
)
except UserNotFound as e:
return UserMutationReturn(
success=False,
message=str(e),
except UserNotFound as error:
return return_failed_mutation_return(
message=error.get_error_message(),
code=404,
)
except SelfPrivacyAppIsOutdate:
return UserMutationReturn(
success=False,
message=f"Error: Failed to change password. {PLEASE_UPDATE_APP_TEXT}",
code=400,
)
return UserMutationReturn(
success=True,
@ -186,28 +179,23 @@ class UsersMutations:
try:
create_ssh_key_action(ssh_input.username, ssh_input.ssh_key)
except KeyAlreadyExists:
return UserMutationReturn(
success=False,
message="Key already exists",
except KeyAlreadyExists as error:
return return_failed_mutation_return(
message=error.get_error_message(),
code=409,
)
except InvalidPublicKey:
return UserMutationReturn(
success=False,
message="Invalid key type. Only ssh-ed25519, ssh-rsa and ecdsa are supported",
code=400,
except InvalidPublicKey as error:
return return_failed_mutation_return(
message=error.get_error_message(),
)
except UserNotFound:
return UserMutationReturn(
success=False,
message="User not found",
except UserNotFound as error:
return return_failed_mutation_return(
message=error.get_error_message(),
code=404,
)
except Exception as e: # TODO why?
return UserMutationReturn(
success=False,
message=str(e),
except Exception as error: # TODO why?
return return_failed_mutation_return(
message=str(error),
code=500,
)
@ -224,22 +212,15 @@ class UsersMutations:
try:
remove_ssh_key_action(ssh_input.username, ssh_input.ssh_key)
except KeyNotFound:
return UserMutationReturn(
success=False,
message="Key not found",
except (KeyNotFound, UserMutationReturn) as error:
return return_failed_mutation_return(
message=error.get_error_message(),
code=404,
)
except UserNotFound:
except Exception as error: # TODO why?
return UserMutationReturn(
success=False,
message="User not found",
code=404,
)
except Exception as e: # TODO why?
return UserMutationReturn(
success=False,
message=str(e),
message=str(error),
code=500,
)
@ -254,10 +235,9 @@ class UsersMutations:
def generate_password_reset_link(username: str) -> UserMutationReturn:
try:
password_reset_link = generate_password_reset_link_action(username=username)
except UserNotFound:
return UserMutationReturn(
success=False,
message="User not found",
except UserNotFound as error:
return return_failed_mutation_return(
message=error.get_error_message(),
code=404,
)

56
selfprivacy_api/repositories/users/exceptions.py
View file

@ -1,34 +1,74 @@
class UserNotFound(Exception):
"""Attemted to get a user that does not exist"""
"""Attempted to get a user that does not exist"""
@staticmethod
def get_error_message() -> str:
return "User not found"
class UserIsProtected(Exception):
"""Attemted to delete a user that is protected"""
"""Attempted to delete a user that is protected"""
@staticmethod
def get_error_message() -> str:
return "User is protected and cannot be deleted"
class UsernameForbidden(Exception):
"""Attemted to create a user with a forbidden username"""
"""Attempted to create a user with a forbidden username"""
@staticmethod
def get_error_message() -> str:
return "Username is forbidden"
class UserAlreadyExists(Exception):
"""Attemted to create a user that already exists"""
"""Attempted to create a user that already exists"""
@staticmethod
def get_error_message() -> str:
return "User already exists"
class UsernameNotAlphanumeric(Exception):
"""Attemted to create a user with a non-alphanumeric username"""
"""Attempted to create a user with a non-alphanumeric username"""
@staticmethod
def get_error_message() -> str:
return "Username not alphanumeric"
class UsernameTooLong(Exception):
"""Attemted to create a user with a too long username. Username must be less than 32 characters"""
"""
Attempted to create a user with a too long username. Username must be less than 32 characters
"""
@staticmethod
def get_error_message() -> str:
return "Username is too long. Must be less than 32 characters"
class PasswordIsEmpty(Exception):
"""Attemted to create a user with an empty password"""
"""Attempted to create a user with an empty password"""
@staticmethod
def get_error_message() -> str:
return "Password cannot be empty"
class InvalidConfiguration(Exception):
"""The userdata is broken"""
@staticmethod
def get_error_message() -> str:
return "Invalid configuration, userdata is broken"
class SelfPrivacyAppIsOutdate(Exception):
"""SelfPrivacy app is out of date, please update. Some important functions are not working at the moment."""
"""
SelfPrivacy app is out of date, please update. Some important functions are not working at the moment.
"""
@staticmethod
def get_error_message() -> str:
return "SelfPrivacy app is out of date, please update"

4
selfprivacy_api/repositories/users/kanidm_user_repository.py
View file

@ -86,6 +86,10 @@ class KanidmAdminToken: # TODO CHECK IS TOKEN CORRECT?
return new_kanidm_admin_password
@staticmethod
def _delete_kanidm_token_from_db() -> None:
redis.delete("kanidm:token")
class KanidmQueryError(Exception):
"""Error occurred during kanidm query"""

10
selfprivacy_api/services/__init__.py
View file

@ -66,7 +66,15 @@ class ServiceManager(Service):
ip4 = network_utils.get_ip4()
ip6 = network_utils.get_ip6()
dns_records: list[ServiceDnsRecord] = []
dns_records: list[ServiceDnsRecord] = [
ServiceDnsRecord(
type="A",
name="auth",
content=ip4,
ttl=3600,
display_name="Record for Kanidm",
),
]
try:
dns_records.append(

4
tests/test_ssh.py
View file

@ -14,12 +14,12 @@ from selfprivacy_api.actions.ssh import (
KeyNotFound,
UserNotFound,
)
from selfprivacy_api.actions.users import (
from selfprivacy_api.repositories.users.json_user_repository import (
get_users,
get_user_by_username,
UserDataUserOrigin,
)
from selfprivacy_api.utils import WriteUserData, ReadUserData
from selfprivacy_api.models.user import UserDataUserOrigin
@pytest.fixture(params=[True, False])