implement renewals and password resets

pull/5/head
Max Erenberg 2 years ago
parent da14764687
commit c32e565f68
  1. 4
      ceo_common/errors.py
  2. 26
      ceod/api/members.py
  3. 8
      ceod/model/User.py
  4. 6
      ceod/model/validators.py
  5. 9
      ceod/transactions/AbstractTransaction.py
  6. 8
      ceod/transactions/members/AddMemberTransaction.py
  7. 42
      ceod/transactions/members/RenewMemberTransaction.py
  8. 27
      ceod/transactions/members/ResetPasswordTransaction.py
  9. 2
      ceod/transactions/members/__init__.py
  10. 7
      ceod/transactions/members/utils.py

@ -4,3 +4,7 @@ class UserNotFoundError(Exception):
class GroupNotFoundError(Exception):
pass
class BadRequest(Exception):
pass

@ -1,13 +1,16 @@
from flask import Blueprint, request
from zope import component
from .utils import authz_restrict_to_staff, user_is_in_group, \
requires_authentication_no_realm, create_streaming_response
from .utils import authz_restrict_to_staff, authz_restrict_to_syscom, \
user_is_in_group, requires_authentication_no_realm, \
create_streaming_response
from ceo_common.errors import UserNotFoundError
from ceo_common.interfaces import ILDAPService
from ceod.transactions.members import (
AddMemberTransaction,
ModifyMemberTransaction,
RenewMemberTransaction,
ResetPasswordTransaction,
)
bp = Blueprint('members', __name__)
@ -57,3 +60,22 @@ def patch_user(auth_user: str, username: str):
forwarding_addresses=body.get('forwarding_addresses'),
)
return create_streaming_response(txn)
@bp.route('/<username>/renew', methods=['POST'])
@authz_restrict_to_staff
def renew_user(username: str):
body = request.get_json(force=True)
txn = RenewMemberTransaction(
username,
terms=body.get('terms'),
non_member_terms=body.get('non_member_terms'),
)
return create_streaming_response(txn)
@bp.route('/<username>/pwreset', methods=['POST'])
@authz_restrict_to_syscom
def reset_user_password(username: str):
txn = ResetPasswordTransaction(username)
return create_streaming_response(txn)

@ -7,7 +7,7 @@ from zope import component
from zope.interface import implementer
from .utils import strings_to_bytes, bytes_to_strings
from .validators import is_valid_shell
from .validators import is_valid_shell, is_valid_term
from ceo_common.interfaces import ILDAPService, IKerberosService, IFileService, \
IUser, IConfig, IMailmanService
@ -178,6 +178,9 @@ class User:
self.ldap_srv.modify_user(self, new_user)
def add_terms(self, terms: List[str]):
for term in terms:
if not is_valid_term(term):
raise Exception('%s is not a valid term' % term)
new_user = copy.copy(self)
new_user.terms = self.terms.copy()
new_user.terms.extend(terms)
@ -185,6 +188,9 @@ class User:
self.terms = new_user.terms
def add_non_member_terms(self, terms: List[str]):
for term in terms:
if not is_valid_term(term):
raise Exception('%s is not a valid term' % term)
new_user = copy.copy(self)
new_user.non_member_terms = self.non_member_terms.copy()
new_user.non_member_terms.extend(terms)

@ -58,3 +58,9 @@ def is_valid_shell(shell: str) -> bool:
line.strip() for line in open('/etc/shells')
if line != '' and not line.isspace()
]
def is_valid_term(term: str) -> bool:
return len(term) == 5 and \
term[0] in ['s', 'f', 'w'] and \
term[1:5].isdigit()

@ -41,7 +41,10 @@ class AbstractTransaction(ABC):
for _ in self.execute_iter():
pass
@abstractmethod
def rollback(self):
"""Roll back the transaction, when it fails."""
raise NotImplementedError()
"""
Roll back the transaction, when it fails.
If the transaction only has one operation, then there is no need
to implement this.
"""
pass

@ -1,11 +1,10 @@
import base64
import os
import traceback
from typing import Union, List
from zope import component
from ..AbstractTransaction import AbstractTransaction
from .utils import gen_password
from ceo_common.interfaces import IConfig, IMailService
from ceo_common.logger_factory import logger_factory
from ceod.model import User, Group
@ -13,11 +12,6 @@ from ceod.model import User, Group
logger = logger_factory(__name__)
def gen_password() -> str:
"""Generate a temporary password."""
return base64.b64encode(os.urandom(18)).decode()
class AddMemberTransaction(AbstractTransaction):
"""Transaction to add a new club member."""

@ -0,0 +1,42 @@
from typing import Union, List
from zope import component
from ..AbstractTransaction import AbstractTransaction
from ceo_common.errors import BadRequest
from ceo_common.interfaces import ILDAPService
class RenewMemberTransaction(AbstractTransaction):
"""Transaction to renew a user's terms or non-member terms."""
operations = [
'add_terms',
'add_non_member_terms',
]
def __init__(
self,
username: str,
terms: Union[List[str], None],
non_member_terms: Union[List[str], None],
):
super().__init__()
self.username = username
if (terms and non_member_terms) or not (terms or non_member_terms):
raise BadRequest('Must specify either terms or non-member terms')
self.terms = terms
self.non_member_terms = non_member_terms
self.ldap_srv = component.getUtility(ILDAPService)
def child_execute_iter(self):
user = self.ldap_srv.get_user(self.username)
if self.terms:
user.add_terms(self.terms)
yield 'add_terms'
elif self.non_member_terms:
user.add_non_member_terms(self.non_member_terms)
yield 'add_non_member_terms'
self.finish('OK')

@ -0,0 +1,27 @@
from zope import component
from ..AbstractTransaction import AbstractTransaction
from .utils import gen_password
from ceo_common.interfaces import ILDAPService
class ResetPasswordTransaction(AbstractTransaction):
"""Transaction to reset a user's password."""
operations = [
'change_password',
]
def __init__(self, username: str):
super().__init__()
self.username = username
self.ldap_srv = component.getUtility(ILDAPService)
def child_execute_iter(self):
user = self.ldap_srv.get_user(self.username)
password = gen_password()
user.change_password(password)
yield 'change_password'
self.finish({'password': password})

@ -1,2 +1,4 @@
from .AddMemberTransaction import AddMemberTransaction
from .ModifyMemberTransaction import ModifyMemberTransaction
from .RenewMemberTransaction import RenewMemberTransaction
from .ResetPasswordTransaction import ResetPasswordTransaction

@ -0,0 +1,7 @@
import base64
import os
def gen_password() -> str:
"""Generate a temporary password."""
return base64.b64encode(os.urandom(18)).decode()
Loading…
Cancel
Save