Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
request = NewAccountRequest(
user_name=account['user_name'],
real_name=account['group_name'],
is_group=True,
calnet_uid=None,
callink_oid=account['callink_oid'],
email=account['email'],
encrypted_password=encrypt_password(
password,
RSA.importKey(CREATE_PUBLIC_KEY),
),
handle_warnings=NewAccountRequest.WARNINGS_WARN,
)
print()
print(bold('Pending account request:'))
print(dedent(
"""\
User Name: {request.user_name}
Group Name: {request.real_name}
CalLink OID: {request.callink_oid}
Email: {request.email}
"""
).format(request=request))
if input('Submit request? [yN] ') != 'y':
input('Press enter to continue.')
continue
conf = ConfigParser()
conf.read('/etc/ocf-create/ocf-create.conf')
_logger.error(format_exc())
except Exception as ex:
print(ex) # just in case it errors again here
send_problem_report(
dedent(
"""\
An exception occured in ocfweb, but we errored trying to report it:
{traceback}
"""
).format(traceback=format_exc()),
)
raise
else:
_logger.debug(bold(yellow(f'Not updating periodic function: {pf}')))
if was_error:
delay_on_error = min(DELAY_ON_ERROR_MAX, delay_on_error * 2)
time.sleep(delay_on_error)
else:
delay_on_error = max(DELAY_ON_ERROR_MIN, delay_on_error / 2)
'modify the request.'
))
choice = input('Create the account anyway? [yN] ')
if choice in ('y', 'Y'):
new_request = request._replace(
handle_warnings=NewAccountRequest.WARNINGS_CREATE,
)
task = tasks.validate_then_create_account.delay(new_request)
response = wait_for_task(celery, task)
else:
input('Starting over, press enter to continue...')
continue
if response.status == NewAccountResponse.CREATED:
print(bold(green('Account created!')))
print('Your account was created successfully.')
print('You\'ve been sent an email with more information.')
return
else:
# this shouldn't be possible; we must have entered some weird state
# TODO: report via ocflib
print(bold(red('Error: Entered unexpected state.')))
print(red('The request we submitted was:'))
print(red(request))
print(red('The new request we submitted (if any) was:'))
print(red(new_request))
print(red('The response we received was:'))
print(red(response))
print(bold(red('Not really sure what to do here, sorry.')))
input('Press enter to start over...')
def run_periodic_functions() -> None:
global delay_on_error
# First, import urls so that views are imported, decorators are run, and
# our periodic functions get registered.
import ocfweb.urls # noqa
was_error = False
for pf in periodic_functions:
if pf.seconds_since_last_update() >= pf.period:
_logger.info(bold(green(f'Updating periodic function: {pf}')))
try:
pf.update()
except Exception as ex:
was_error = True
if isinstance(ex, KeyboardInterrupt) or settings.DEBUG:
raise
try:
send_problem_report(
dedent(
"""\
An exception occurred in an ocfweb periodic function:
{traceback}
if response.status == NewAccountResponse.CREATED:
print(bold(green('Account created!')))
print('Your account was created successfully.')
print('You\'ve been sent an email with more information.')
return
else:
# this shouldn't be possible; we must have entered some weird state
# TODO: report via ocflib
print(bold(red('Error: Entered unexpected state.')))
print(red('The request we submitted was:'))
print(red(request))
print(red('The new request we submitted (if any) was:'))
print(red(new_request))
print(red('The response we received was:'))
print(red(response))
print(bold(red('Not really sure what to do here, sorry.')))
input('Press enter to start over...')
tasks = get_tasks(celery)
task = tasks.validate_then_create_account.delay(request)
response = wait_for_task(celery, task)
new_request = None
if response.status == NewAccountResponse.REJECTED:
print(bold(red(
'Account requested was rejected for the following reasons:'
)))
for error in response.errors:
print(red(' - {}'.format(error)))
input('Press enter to start over (or ^C to cancel)...')
continue
elif response.status == NewAccountResponse.FLAGGED:
print(bold(yellow(
'Account requested was flagged for the following reasons:'
)))
for error in response.errors:
print(yellow(' - {}'.format(error)))
print(bold(
'You can either create the account anyway, or go back and '
'modify the request.'
))
choice = input('Create the account anyway? [yN] ')
if choice in ('y', 'Y'):
new_request = request._replace(
handle_warnings=NewAccountRequest.WARNINGS_CREATE,
)
task = tasks.validate_then_create_account.delay(new_request)
response = wait_for_task(celery, task)
if response.status == NewAccountResponse.REJECTED:
print(bold(red(
'Account requested was rejected for the following reasons:'
)))
for error in response.errors:
print(red(' - {}'.format(error)))
input('Press enter to start over (or ^C to cancel)...')
continue
elif response.status == NewAccountResponse.FLAGGED:
print(bold(yellow(
'Account requested was flagged for the following reasons:'
)))
for error in response.errors:
print(yellow(' - {}'.format(error)))
print(bold(
'You can either create the account anyway, or go back and '
'modify the request.'
))
choice = input('Create the account anyway? [yN] ')
if choice in ('y', 'Y'):
new_request = request._replace(
handle_warnings=NewAccountRequest.WARNINGS_CREATE,
)
task = tasks.validate_then_create_account.delay(new_request)
response = wait_for_task(celery, task)
else:
input('Starting over, press enter to continue...')
continue
if response.status == NewAccountResponse.CREATED: