Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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}
Periodic function:
* Key: {pf.function_call_key}
* Last Update: {last_update} ({seconds_since_last_update} seconds ago)
* Period: {pf.period}
* TTL: {pf.ttl}
The background process will now pause for {delay} seconds.
"""
).format(
traceback=format_exc(),
def username_reserved(username):
if username.startswith('ocf'):
return True
if username in RESERVED_USERNAMES:
return True
# sanity check: make sure no local users share the username
with open('/etc/passwd') as f:
if any(line.startswith(username + ':') for line in f):
print(
'WARNING: Username {} rejected based on /etc/passwd!'
.format(username),
file=sys.stderr)
ocflib.misc.mail.send_problem_report(
"""Username {} rejected based on /etc/passwd. It should be \
added to RESERVED_USERNAMES for consistency across \
servers!""".format(username))
return True
return False
def handle_error(self, error_message):
# for debugging purposes
print(error_message)
# don't send emails when running as dev
if not TESTING:
send_problem_report(error_message)
command = ('/usr/bin/ldapmodify', '-Q')
try:
subprocess.check_output(
command,
input='\n'.join(lines),
universal_newlines=True,
timeout=10,
)
except subprocess.CalledProcessError as e:
if e.returncode == 32:
raise ValueError('Tried to modify nonexistent entry.')
elif e.returncode == 68:
raise ValueError('Tried to create duplicate entry.')
else:
send_problem_report(
dedent(
'''\
Unknown problem occured when trying to write to LDAP; the
code should be updated to handle this case.
dn: {dn}
keytab: {keytab}
principal: {principal}
Error code: {returncode}
Unexpected output:
{output}
Lines passed to ldapmodify:
{lines}
Session:
{session}
"""
).format(
traceback=traceback,
request=request,
host=request.get_host(),
path=request.get_full_path(),
is_secure=request.is_secure(),
session=pformat(dict(request.session)),
headers=pformat(headers),
),
)
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
command = ('/usr/bin/ldapmodify', '-Q')
try:
subprocess.check_output(
command,
input='\n'.join(lines),
universal_newlines=True,
timeout=10,
)
except subprocess.CalledProcessError as e:
if e.returncode == 32:
raise ValueError('Tried to modify nonexistent entry.')
elif e.returncode == 68:
raise ValueError('Tried to create duplicate entry.')
else:
send_problem_report(
dedent(
'''\
Unknown problem occured when trying to write to LDAP; the
code should be updated to handle this case.
dn: {dn}
keytab: {keytab}
principal: {principal}
Error code: {returncode}
Unexpected output:
{output}
Lines passed to ldapmodify:
{lines}
def process_exception(self, request, exception):
if settings.DEBUG:
return
try:
send_problem_report(dedent(
"""\
An exception occured in atool:
{traceback}
Request:
* Host: {host}
* Path: {path}
* Method: {request.method}
* Secure: {is_secure}
Request Headers:
{headers}
Session: