policy(minor): Move policy-related code to separate package
This commit is contained in:
		| @ -4,7 +4,7 @@ from datetime import timedelta | ||||
| from logging import getLogger | ||||
| from random import SystemRandom | ||||
| from time import sleep | ||||
| from typing import Tuple, Union | ||||
| from typing import List | ||||
| from uuid import uuid4 | ||||
|  | ||||
| from django.contrib.auth.models import AbstractUser | ||||
| @ -25,6 +25,20 @@ def default_nonce_duration(): | ||||
|     """Default duration a Nonce is valid""" | ||||
|     return now() + timedelta(hours=4) | ||||
|  | ||||
|  | ||||
| class PolicyResult: | ||||
|     """Small data-class to hold policy results""" | ||||
|  | ||||
|     passing: bool = False | ||||
|     messages: List[str] = [] | ||||
|  | ||||
|     def __init__(self, passing: bool, *messages: str): | ||||
|         self.passing = passing | ||||
|         self.messages = messages | ||||
|  | ||||
|     def __str__(self): | ||||
|         return f"<PolicyResult passing={self.passing}>" | ||||
|  | ||||
| class Group(UUIDModel): | ||||
|     """Custom Group model which supports a basic hierarchy""" | ||||
|  | ||||
| @ -229,7 +243,7 @@ class Policy(UUIDModel, CreatedUpdatedModel): | ||||
|             return self.name | ||||
|         return "%s action %s" % (self.name, self.action) | ||||
|  | ||||
|     def passes(self, user: User) -> Union[bool, Tuple[bool, str]]: | ||||
|     def passes(self, user: User) -> PolicyResult: | ||||
|         """Check if user instance passes this policy""" | ||||
|         raise NotImplementedError() | ||||
|  | ||||
| @ -273,7 +287,7 @@ class FieldMatcherPolicy(Policy): | ||||
|             description = "%s: %s" % (self.name, description) | ||||
|         return description | ||||
|  | ||||
|     def passes(self, user: User) -> Union[bool, Tuple[bool, str]]: | ||||
|     def passes(self, user: User) -> PolicyResult: | ||||
|         """Check if user instance passes this role""" | ||||
|         if not hasattr(user, self.user_field): | ||||
|             raise ValueError("Field does not exist") | ||||
| @ -294,7 +308,7 @@ class FieldMatcherPolicy(Policy): | ||||
|             passes = user_field_value == self.value | ||||
|  | ||||
|         LOGGER.debug("User got '%r'", passes) | ||||
|         return passes | ||||
|         return PolicyResult(passes) | ||||
|  | ||||
|     class Meta: | ||||
|  | ||||
| @ -313,10 +327,10 @@ class PasswordPolicy(Policy): | ||||
|  | ||||
|     form = 'passbook.core.forms.policies.PasswordPolicyForm' | ||||
|  | ||||
|     def passes(self, user: User) -> Union[bool, Tuple[bool, str]]: | ||||
|     def passes(self, user: User) -> PolicyResult: | ||||
|         # Only check if password is being set | ||||
|         if not hasattr(user, '__password__'): | ||||
|             return True | ||||
|             return PolicyResult(True) | ||||
|         password = getattr(user, '__password__') | ||||
|  | ||||
|         filter_regex = r'' | ||||
| @ -329,8 +343,8 @@ class PasswordPolicy(Policy): | ||||
|         result = bool(re.compile(filter_regex).match(password)) | ||||
|         LOGGER.debug("User got %r", result) | ||||
|         if not result: | ||||
|             return result, self.error_message | ||||
|         return result | ||||
|             return PolicyResult(result, self.error_message) | ||||
|         return PolicyResult(result) | ||||
|  | ||||
|     class Meta: | ||||
|  | ||||
| @ -364,7 +378,7 @@ class WebhookPolicy(Policy): | ||||
|  | ||||
|     form = 'passbook.core.forms.policies.WebhookPolicyForm' | ||||
|  | ||||
|     def passes(self, user: User): | ||||
|     def passes(self, user: User) -> PolicyResult: | ||||
|         """Call webhook asynchronously and report back""" | ||||
|         raise NotImplementedError() | ||||
|  | ||||
| @ -383,12 +397,12 @@ class DebugPolicy(Policy): | ||||
|  | ||||
|     form = 'passbook.core.forms.policies.DebugPolicyForm' | ||||
|  | ||||
|     def passes(self, user: User): | ||||
|     def passes(self, user: User) -> PolicyResult: | ||||
|         """Wait random time then return result""" | ||||
|         wait = SystemRandom().randrange(self.wait_min, self.wait_max) | ||||
|         LOGGER.debug("Policy '%s' waiting for %ds", self.name, wait) | ||||
|         sleep(wait) | ||||
|         return self.result, 'Debugging' | ||||
|         return PolicyResult(self.result, 'Debugging') | ||||
|  | ||||
|     class Meta: | ||||
|  | ||||
| @ -402,8 +416,8 @@ class GroupMembershipPolicy(Policy): | ||||
|  | ||||
|     form = 'passbook.core.forms.policies.GroupMembershipPolicyForm' | ||||
|  | ||||
|     def passes(self, user: User) -> Union[bool, Tuple[bool, str]]: | ||||
|         return self.group.user_set.filter(pk=user.pk).exists() | ||||
|     def passes(self, user: User) -> PolicyResult: | ||||
|         return PolicyResult(self.group.user_set.filter(pk=user.pk).exists()) | ||||
|  | ||||
|     class Meta: | ||||
|  | ||||
| @ -415,10 +429,10 @@ class SSOLoginPolicy(Policy): | ||||
|  | ||||
|     form = 'passbook.core.forms.policies.SSOLoginPolicyForm' | ||||
|  | ||||
|     def passes(self, user): | ||||
|     def passes(self, user) -> PolicyResult: | ||||
|         """Check if user instance passes this policy""" | ||||
|         from passbook.core.auth.view import AuthenticationView | ||||
|         return user.session.get(AuthenticationView.SESSION_IS_SSO_LOGIN, False), "" | ||||
|         return PolicyResult(user.session.get(AuthenticationView.SESSION_IS_SSO_LOGIN, False)) | ||||
|  | ||||
|     class Meta: | ||||
|  | ||||
|  | ||||
		Reference in New Issue
	
	Block a user
	 Langhammer, Jens
					Langhammer, Jens