gauvainboiche
68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
from domain.exceptions import (
|
|
AccessDeniedError,
|
|
AuthenticationError,
|
|
ConflictError,
|
|
SecretNotFoundError,
|
|
)
|
|
from domain.models.secrets import Secret
|
|
from domain.models.users import User
|
|
|
|
def authenticate_user(username: str, password: str, user_repository, crypto) -> User:
|
|
user = user_repository.find_by_username(username)
|
|
if user is None or not user.is_active:
|
|
raise AuthenticationError("Identifiants incorrects.")
|
|
if not crypto.verify_password(password, user.password_hash):
|
|
raise AuthenticationError("Identifiants incorrects.")
|
|
return user
|
|
|
|
def check_team_membership(user: User, team_id: int) -> None:
|
|
if team_id not in user.teams_id:
|
|
raise AccessDeniedError("Vous n'\u00eates pas membre de cette \u00e9quipe.")
|
|
|
|
def list_secrets_for_team(user: User, team_id: int, secret_repository) -> list[Secret]:
|
|
check_team_membership(user, team_id)
|
|
return secret_repository.find_by_team_id(team_id)
|
|
|
|
def reveal_secret(user: User, secret_id: int, secret_repository, crypto) -> str:
|
|
secret = secret_repository.find_by_id(secret_id)
|
|
if secret is None:
|
|
raise SecretNotFoundError("Secret introuvable.")
|
|
check_team_membership(user, secret.team_id)
|
|
return crypto.decrypt_secret(secret.encrypted_value)
|
|
|
|
def create_secret(
|
|
user: User,
|
|
team_id: int,
|
|
name: str,
|
|
value: str,
|
|
secret_repository,
|
|
crypto,
|
|
) -> Secret:
|
|
check_team_membership(user, team_id)
|
|
encrypted_value = crypto.encrypt_secret(value)
|
|
return secret_repository.create(name=name, encrypted_value=encrypted_value, team_id=team_id)
|
|
|
|
def rotate_secret(
|
|
user: User,
|
|
secret_id: int,
|
|
new_value: str,
|
|
expected_version: int,
|
|
secret_repository,
|
|
crypto,
|
|
) -> Secret:
|
|
secret = secret_repository.find_by_id(secret_id)
|
|
if secret is None:
|
|
raise SecretNotFoundError("Secret introuvable.")
|
|
check_team_membership(user, secret.team_id)
|
|
if secret.version != expected_version:
|
|
raise ConflictError(
|
|
"Ce secret a été modifié par quelqu'un d'autre depuis que vous avez ouvert "
|
|
"le formulaire. Veuillez recharger la page et réessayer."
|
|
)
|
|
encrypted_value = crypto.encrypt_secret(new_value)
|
|
updated = secret_repository.update(secret_id, encrypted_value, expected_version)
|
|
if updated is None:
|
|
raise ConflictError(
|
|
"Conflit de modification détecté. Veuillez recharger et réessayer."
|
|
)
|
|
return updated |