"""Miscellaneous functions for the mudpy engine."""
-# Copyright (c) 2004-2019 mudpy authors. Permission to use, copy,
+# Copyright (c) 2004-2020 mudpy authors. Permission to use, copy,
# modify, and distribute this software is granted under terms
# provided in the LICENSE file distributed with this software.
prepend_padding
)
+ def is_admin(self):
+ """Boolean check whether an actor is controlled by an admin owner."""
+ return(self.owner and self.owner.is_admin())
+
def can_run(self, command):
"""Check if the user can run this command object."""
# has to be in the commands group
if command not in self.universe.groups["command"].values():
- result = False
+ return(False)
# avatars of administrators can run any command
- elif self.owner and self.owner.account.get("administrator"):
- result = True
+ if self.is_admin():
+ return(True)
# everyone can run non-administrative commands
- elif not command.get("administrative"):
- result = True
+ if not command.get("administrative"):
+ return(True)
# otherwise the command cannot be run by this actor
- else:
- result = False
-
- # pass back the result
- return result
+ return(False)
def update_location(self):
"""Make sure the location's contents contain this element."""
self.output_queue = []
self.partial_input = b""
self.password_tries = 0
+ self.rows = 23
self.state = "telopt_negotiation"
self.telopts = {}
+ self.ttype = None
self.universe = universe
def quit(self):
else:
self.check_idle()
+ # ask the client for their current terminal type (RFC 1091); it's None
+ # if it's not been initialized, the empty string if it has but the
+ # output was indeterminate, "UNKNOWN" if the client specified it has no
+ # terminal types to supply
+ if self.ttype is None:
+ mudpy.telnet.request_ttype(self)
+
# if output is paused, decrement the counter
if self.state == "telopt_negotiation":
if self.negotiation_pause:
# check for some input
try:
raw_input = self.connection.recv(1024)
- except (BlockingIOError, OSError):
+ except OSError:
raw_input = b""
# we got something
avatar, 6)
return avatars
+ def is_admin(self):
+ """Boolean check whether user's account is an admin."""
+ return(self.account.get("administrator", False))
+
def broadcast(message, add_prompt=True):
"""Send a message to all connected users."""
# display to connected administrators
for user in universe.userlist:
- if user.state == "active" and user.account.get(
- "administrator"
- ) and user.account.get("loglevel", 0) <= level:
+ if (
+ user.state == "active"
+ and user.is_admin()
+ and user.account.get("loglevel", 0) <= level):
# iterate over every line in the message
full_message = ""
for line in lines:
# don't proceed if there are no lines
if filtered_count:
- # can't start before the begining or at the end
+ # can't start before the beginning or at the end
if start > filtered_count:
start = filtered_count
if start < 1:
# characters, printable or otherwise
abs_pos = 0
- # the current text position relative to the begining of the line,
+ # the current text position relative to the beginning of the line,
# ignoring color escape sequences
rel_pos = 0
# normalize any potentially composited unicode before we count it
text = unicodedata.normalize("NFKC", text)
- # iterate over each character from the begining of the text
+ # iterate over each character from the beginning of the text
for each_character in text:
# the current character is the escape character
# characters but the space it replaced was only one
abs_pos += 1
- # now we're at the begining of a new line, plus the
+ # now we're at the beginning of a new line, plus the
# number of characters wrapped from the previous line
rel_pos -= last_rel_whitespace
last_rel_whitespace = 0
expanded.append(key)
# return one at random
- return random.choice(expanded)
+ # Allow the random.randrange() call in bandit since it's not used for
+ # security/cryptographic purposes
+ return random.choice(expanded) # nosec
def random_name():
name = ""
# create a name of random length from the syllables
- for _syllable in range(random.randrange(2, 6)):
+ # Allow the random.randrange() call in bandit since it's not used for
+ # security/cryptographic purposes
+ for _syllable in range(random.randrange(2, 6)): # nosec
name += weighted_choice(syllables)
# strip any leading quotemark, capitalize and return the name
"""Reload all relevant objects."""
universe.save()
old_userlist = universe.userlist[:]
+ old_loglines = universe.loglines[:]
for element in list(universe.contents.values()):
element.destroy()
universe.load()
+ new_loglines = universe.loglines[:]
+ universe.loglines = old_loglines + new_loglines
for user in old_userlist:
user.reload()
def call_hook_function(fname, arglist):
"""Safely execute named function with supplied arguments, return result."""
- # strip any explicit leader or parameter
- # TODO(fungi) remove this once the menu functions transition is complete
- if fname.startswith("mudpy."):
- fname = fname[6:]
- if fname.endswith("(user)"):
- fname = fname[:-6]
-
# all functions relative to mudpy package
function = mudpy
command = find_command(command_name)
# if it's allowed, do it
- ran = False
+ result = None
if actor.can_run(command):
- # dereference the relative object path for the requested function
- # TODO(fungi) use call_hook_function() here instead
- action = mudpy
action_fname = command.get("action", command.key)
- for component in action_fname.split("."):
- try:
- action = getattr(action, component)
- ran = True
- except AttributeError:
- log('Could not find action function "%s" for command "%s"'
- % (action_fname, command_name))
- action = None
- break
- if action:
- try:
- action(actor, parameters)
- except Exception:
- log('Command string "%s" from user %s raised an '
- 'exception...\n%s' % (
- input_data, actor.owner.account.get("name"),
- traceback.format_exc()))
- mudpy.command.error(actor, input_data)
+ if action_fname:
+ result = call_hook_function(action_fname, (actor, parameters))
# if the command was not run, give an error
- if not ran:
+ if not result:
mudpy.command.error(actor, input_data)
# if no input, just idle back with a prompt
projects / mudpy.git / blobdiff