"""Core objects for the mudpy engine."""
-# Copyright (c) 2005 mudpy, Jeremy Stanley <fungi@yuggoth.org>, all rights reserved.
+# Copyright (c) 2006 mudpy, Jeremy Stanley <fungi@yuggoth.org>, all rights reserved.
# Licensed per terms in the LICENSE file distributed with this software.
# import some things we need
-from ConfigParser import SafeConfigParser
+from ConfigParser import RawConfigParser
from md5 import new as new_md5
-from os import F_OK, R_OK, access, getcwd, makedirs, sep
+from os import _exit, R_OK, W_OK, access, chmod, close, fork, getpid, makedirs, remove, setsid, stat
+from os.path import abspath, dirname, exists, isabs, join as path_join
from random import choice, randrange
+from re import match
+from signal import SIGHUP, SIGTERM, signal
from socket import AF_INET, SO_REUSEADDR, SOCK_STREAM, SOL_SOCKET, socket
+from stat import S_IMODE, ST_MODE
+from sys import stderr
+from syslog import LOG_PID, LOG_INFO, LOG_DAEMON, closelog, openlog, syslog
+from telnetlib import DO, DONT, ECHO, EOR, GA, IAC, LINEMODE, SB, SE, SGA, WILL, WONT
from time import asctime, sleep
+from traceback import format_exception
-# a dict of replacement macros
-macros = {
- "$(eol)": "\r\n",
- "$(bld)": chr(27) + "[1m",
- "$(nrm)": chr(27) + "[0m",
- "$(blk)": chr(27) + "[30m",
- "$(grn)": chr(27) + "[32m",
- "$(red)": chr(27) + "[31m"
- }
+def excepthook(excepttype, value, traceback):
+ """Handle uncaught exceptions."""
+
+ # assemble the list of errors into a single string
+ message = "".join(format_exception(excepttype, value, traceback))
+
+ # try to log it, if possible
+ try: log(message, 9)
+ except: pass
+
+ # try to write it to stderr, if possible
+ try: stderr.write(message)
+ except: pass
+
+# redefine sys.excepthook with ours
+import sys
+sys.excepthook = excepthook
+
+def sighook(what, where):
+ """Handle external signals."""
+
+ # a generic message
+ message = "Caught signal: "
+
+ # for a hangup signal
+ if what == SIGHUP:
+ message += "hangup (reloading)"
+ universe.reload_modules = True
+
+ # for a terminate signal
+ elif what == SIGTERM:
+ message += "terminate (halting)"
+ universe.terminate_world = True
+
+ # catchall for unexpected signals
+ else: message += str(what) + " (unhandled)"
+
+ # log what happened
+ log(message, 8)
+
+# assign the sgnal handlers
+signal(SIGHUP, sighook)
+signal(SIGTERM, sighook)
class Element:
"""An element of the universe."""
- def __init__(self, key, universe, origin=""):
- """Default values for the in-memory element variables."""
+ def __init__(self, key, universe, filename=None):
+ """Set up a new element."""
+
+ # keep track of our key name
self.key = key
- if self.key.find(":") > 0:
- self.category, self.subkey = self.key.split(":", 1)
+
+ # keep track of what universe it's loading` into
+ self.universe = universe
+
+ # clone attributes if this is replacing another element
+ if self.key in self.universe.contents:
+ old_element = self.universe.contents[self.key]
+ for attribute in vars(old_element).keys():
+ exec("self." + attribute + " = old_element." + attribute)
+ if self.owner: self.owner.avatar = self
+
+ # i guess this is a new element then
else:
- self.category = "other"
- self.subkey = self.key
- if not self.category in universe.categories: self.category = "other"
- universe.categories[self.category][self.subkey] = self
- self.origin = origin
- if not self.origin: self.origin = universe.default_origins[self.category]
- if not self.origin.startswith(sep):
- self.origin = getcwd() + sep + self.origin
- universe.contents[self.key] = self
- if not self.origin in universe.files:
- DataFile(self.origin, universe)
- if not universe.files[self.origin].data.has_section(self.key):
- universe.files[self.origin].data.add_section(self.key)
- def delete(self):
- log("Deleting: " + self.key + ".")
- universe.files[self.origin].data.remove_section(self.key)
- del universe.categories[self.category][self.subkey]
- del universe.contents[self.key]
+
+ # not owned by a user by default (used for avatars)
+ self.owner = None
+
+ # no contents in here by default
+ self.contents = {}
+
+ # parse out appropriate category and subkey names, add to list
+ if self.key.find(":") > 0:
+ self.category, self.subkey = self.key.split(":", 1)
+ else:
+ self.category = "other"
+ self.subkey = self.key
+ if not self.category in self.universe.categories:
+ self.category = "other"
+ self.subkey = self.key
+
+ # get an appropriate filename for the origin
+ if not filename: filename = self.universe.default_origins[self.category]
+ if not isabs(filename): filename = abspath(filename)
+
+ # add the file if it doesn't exist yet
+ if not filename in self.universe.files: DataFile(filename, self.universe)
+
+ # record or reset a pointer to the origin file
+ self.origin = self.universe.files[filename]
+
+ # add a data section to the origin if necessary
+ if not self.origin.data.has_section(self.key):
+ self.origin.data.add_section(self.key)
+
+ # add or replace this element in the universe
+ self.universe.contents[self.key] = self
+ self.universe.categories[self.category][self.subkey] = self
+
+ def reload(self):
+ """Create a new element and replace this one."""
+ new_element = Element(self.key, self.universe, self.origin.filename)
+ del(self)
+ def destroy(self):
+ """Remove an element from the universe and destroy it."""
+ self.origin.data.remove_section(self.key)
+ del self.universe.categories[self.category][self.subkey]
+ del self.universe.contents[self.key]
del self
def facets(self):
- """Return a list of facets for this element."""
- return universe.files[self.origin].data.options(self.key)
- def get(self, facet):
+ """Return a list of non-inherited facets for this element."""
+ if self.key in self.origin.data.sections():
+ return self.origin.data.options(self.key)
+ else: return []
+ def has_facet(self, facet):
+ """Return whether the non-inherited facet exists."""
+ return facet in self.facets()
+ def remove_facet(self, facet):
+ """Remove a facet from the element."""
+ if self.has_facet(facet):
+ self.origin.data.remove_option(self.key, facet)
+ self.origin.modified = True
+ def ancestry(self):
+ """Return a list of the element's inheritance lineage."""
+ if self.has_facet("inherit"):
+ ancestry = self.getlist("inherit")
+ for parent in ancestry[:]:
+ ancestors = self.universe.contents[parent].ancestry()
+ for ancestor in ancestors:
+ if ancestor not in ancestry: ancestry.append(ancestor)
+ return ancestry
+ else: return []
+ def get(self, facet, default=None):
"""Retrieve values."""
- if universe.files[self.origin].data.has_option(self.key, facet):
- return universe.files[self.origin].data.get(self.key, facet)
- else:
- return ""
- def getboolean(self, facet, default=False):
+ if default is None: default = ""
+ if self.origin.data.has_option(self.key, facet):
+ return self.origin.data.get(self.key, facet)
+ elif self.has_facet("inherit"):
+ for ancestor in self.ancestry():
+ if self.universe.contents[ancestor].has_facet(facet):
+ return self.universe.contents[ancestor].get(facet)
+ else: return default
+ def getboolean(self, facet, default=None):
"""Retrieve values as boolean type."""
- if universe.files[self.origin].data.has_option(self.key, facet):
- return universe.files[self.origin].data.getboolean(self.key, facet)
- else:
- return default
- def getint(self, facet):
- """Convenience method to coerce return values as type int."""
+ if default is None: default=False
+ if self.origin.data.has_option(self.key, facet):
+ return self.origin.data.getboolean(self.key, facet)
+ elif self.has_facet("inherit"):
+ for ancestor in self.ancestry():
+ if self.universe.contents[ancestor].has_facet(facet):
+ return self.universe.contents[ancestor].getboolean(facet)
+ else: return default
+ def getint(self, facet, default=None):
+ """Return values as int/long type."""
+ if default is None: default = 0
+ if self.origin.data.has_option(self.key, facet):
+ return self.origin.data.getint(self.key, facet)
+ elif self.has_facet("inherit"):
+ for ancestor in self.ancestry():
+ if self.universe.contents[ancestor].has_facet(facet):
+ return self.universe.contents[ancestor].getint(facet)
+ else: return default
+ def getfloat(self, facet, default=None):
+ """Return values as float type."""
+ if default is None: default = 0.0
+ if self.origin.data.has_option(self.key, facet):
+ return self.origin.data.getfloat(self.key, facet)
+ elif self.has_facet("inherit"):
+ for ancestor in self.ancestry():
+ if self.universe.contents[ancestor].has_facet(facet):
+ return self.universe.contents[ancestor].getfloat(facet)
+ else: return default
+ def getlist(self, facet, default=None):
+ """Return values as list type."""
+ if default is None: default = []
value = self.get(facet)
- if not value: value = 0
- elif type(value) is str: value = value.rstrip("L")
- return int(value)
- def getfloat(self, facet):
- """Convenience method to coerce return values as type float."""
+ if value: return makelist(value)
+ else: return default
+ def getdict(self, facet, default=None):
+ """Return values as dict type."""
+ if default is None: default = {}
value = self.get(facet)
- if not value: value = 0
- elif type(value) is str: value = value.rstrip("L")
- return float(value)
+ if value: return makedict(value)
+ else: return default
def set(self, facet, value):
"""Set values."""
- if not type(value) is str: value = repr(value)
- universe.files[self.origin].data.set(self.key, facet, value)
+ if not self.has_facet(facet) or not self.get(facet) == value:
+ if type(value) is long: value = str(value)
+ elif not type(value) is str: value = repr(value)
+ self.origin.data.set(self.key, facet, value)
+ self.origin.modified = True
+ def append(self, facet, value):
+ """Append value tp a list."""
+ if type(value) is long: value = str(value)
+ elif not type(value) is str: value = repr(value)
+ newlist = self.getlist(facet)
+ newlist.append(value)
+ self.set(facet, newlist)
+
+ def new_event(self, action, when=None):
+ """Create, attach and enqueue an event element."""
+
+ # if when isn't specified, that means now
+ if not when: when = self.universe.get_time()
+
+ # events are elements themselves
+ event = Element("event:" + self.key + ":" + counter)
+
+ def send(self, message, eol="$(eol)"):
+ """Convenience method to pass messages to an owner."""
+ if self.owner: self.owner.send(message, eol)
+
+ def can_run(self, command):
+ """Check if the user can run this command object."""
+
+ # has to be in the commands category
+ if command not in self.universe.categories["command"].values(): result = False
+
+ # avatars of administrators can run any command
+ elif self.owner and self.owner.account.getboolean("administrator"): result = True
+
+ # everyone can run non-administrative commands
+ elif not command.getboolean("administrative"): result = True
+
+ # otherwise the command cannot be run by this actor
+ else: result = False
+
+ # pass back the result
+ return result
+
+ def update_location(self):
+ """Make sure the location's contents contain this element."""
+ location = self.get("location")
+ if location in self.universe.contents:
+ self.universe.contents[location].contents[self.key] = self
+ def clean_contents(self):
+ """Make sure the element's contents aren't bogus."""
+ for element in self.contents.values():
+ if element.get("location") != self.key:
+ del self.contents[element.key]
+ def go_to(self, location):
+ """Relocate the element to a specific location."""
+ current = self.get("location")
+ if current and self.key in self.universe.contents[current].contents:
+ del universe.contents[current].contents[self.key]
+ if location in self.universe.contents: self.set("location", location)
+ self.universe.contents[location].contents[self.key] = self
+ self.look_at(location)
+ def go_home(self):
+ """Relocate the element to its default location."""
+ self.go_to(self.get("default_location"))
+ self.echo_to_location("You suddenly realize that " + self.get("name") + " is here.")
+ def move_direction(self, direction):
+ """Relocate the element in a specified direction."""
+ self.echo_to_location(self.get("name") + " exits " + self.universe.categories["internal"]["directions"].getdict(direction)["exit"] + ".")
+ self.send("You exit " + self.universe.categories["internal"]["directions"].getdict(direction)["exit"] + ".")
+ self.go_to(self.universe.contents[self.get("location")].link_neighbor(direction))
+ self.echo_to_location(self.get("name") + " arrives from " + self.universe.categories["internal"]["directions"].getdict(direction)["enter"] + ".")
+ def look_at(self, key):
+ """Show an element to another element."""
+ if self.owner:
+ element = self.universe.contents[key]
+ message = ""
+ name = element.get("name")
+ if name: message += "$(cyn)" + name + "$(nrm)$(eol)"
+ description = element.get("description")
+ if description: message += description + "$(eol)"
+ portal_list = element.portals().keys()
+ if portal_list:
+ portal_list.sort()
+ message += "$(cyn)[ Exits: " + ", ".join(portal_list) + " ]$(nrm)$(eol)"
+ for element in self.universe.contents[self.get("location")].contents.values():
+ if element.getboolean("is_actor") and element is not self:
+ message += "$(yel)" + element.get("name") + " is here.$(nrm)$(eol)"
+ elif element is not self:
+ message += "$(grn)" + element.get("impression") + "$(nrm)$(eol)"
+ self.send(message)
+ def portals(self):
+ """Map the portal directions for a room to neighbors."""
+ portals = {}
+ if match("""^location:-?\d+,-?\d+,-?\d+$""", self.key):
+ coordinates = [(int(x)) for x in self.key.split(":")[1].split(",")]
+ directions = self.universe.categories["internal"]["directions"]
+ offsets = dict([(x, directions.getdict(x)["vector"]) for x in directions.facets()])
+ for portal in self.getlist("gridlinks"):
+ adjacent = map(lambda c,o: c+o, coordinates, offsets[portal])
+ neighbor = "location:" + ",".join([(str(x)) for x in adjacent])
+ if neighbor in self.universe.contents: portals[portal] = neighbor
+ for facet in self.facets():
+ if facet.startswith("link_"):
+ neighbor = self.get(facet)
+ if neighbor in self.universe.contents:
+ portal = facet.split("_")[1]
+ portals[portal] = neighbor
+ return portals
+ def link_neighbor(self, direction):
+ """Return the element linked in a given direction."""
+ portals = self.portals()
+ if direction in portals: return portals[direction]
+ def echo_to_location(self, message):
+ """Show a message to other elements in the current location."""
+ for element in self.universe.contents[self.get("location")].contents.values():
+ if element is not self: element.send(message)
class DataFile:
"""A file containing universe elements."""
def __init__(self, filename, universe):
- filedir = sep.join(filename.split(sep)[:-1])
- self.data = SafeConfigParser()
- if access(filename, R_OK): self.data.read(filename)
self.filename = filename
- universe.files[filename] = self
- if "categories" in self.data.sections():
- for option in self.data.options("categories"):
- universe.default_origins[option] = self.data.get("categories", option)
- if not option in universe.categories:
- universe.categories[option] = {}
+ self.universe = universe
+ self.load()
+ def load(self):
+ """Read a file and create elements accordingly."""
+ self.modified = False
+ self.data = RawConfigParser()
+ if access(self.filename, R_OK): self.data.read(self.filename)
+ if not hasattr(self.universe, "files"): self.universe.files = {}
+ self.universe.files[self.filename] = self
+ if self.data.has_option("__control__", "include_files"):
+ includes = makelist(self.data.get("__control__", "include_files"))
+ else: includes = []
+ if self.data.has_option("__control__", "default_files"):
+ origins = makedict(self.data.get("__control__", "default_files"))
+ for key in origins.keys():
+ if not key in includes: includes.append(key)
+ self.universe.default_origins[key] = origins[key]
+ if not key in self.universe.categories:
+ self.universe.categories[key] = {}
+ if self.data.has_option("__control__", "private_files"):
+ for item in makelist(self.data.get("__control__", "private_files")):
+ if not item in includes: includes.append(item)
+ if not item in self.universe.private_files:
+ if not isabs(item):
+ item = path_join(dirname(self.filename), item)
+ self.universe.private_files.append(item)
for section in self.data.sections():
- if section == "categories" or section == "include":
- for option in self.data.options(section):
- includefile = self.data.get(section, option)
- if not includefile.startswith(sep):
- includefile = filedir + sep + includefile
- DataFile(includefile, universe)
- elif section != "control":
- Element(section, universe, filename)
+ if section != "__control__":
+ Element(section, self.universe, self.filename)
+ for include_file in includes:
+ if not isabs(include_file):
+ include_file = path_join(dirname(self.filename), include_file)
+ if include_file not in self.universe.files or not self.universe.files[include_file].is_writeable():
+ DataFile(include_file, self.universe)
def save(self):
- if self.data.sections() and not ( "control" in self.data.sections() and self.data.getboolean("control", "read_only") ):
- basedir = sep.join(self.filename.split(sep)[:-1])
- if not access(basedir, F_OK): makedirs(basedir)
+ """Write the data, if necessary."""
+
+ # when modified, writeable and has content or the file exists
+ if self.modified and self.is_writeable() and ( self.data.sections() or exists(self.filename) ):
+
+ # make parent directories if necessary
+ if not exists(dirname(self.filename)):
+ makedirs(dirname(self.filename))
+
+ # our data file
file_descriptor = file(self.filename, "w")
- self.data.write(file_descriptor)
+
+ # if it's marked private, chmod it appropriately
+ if self.filename in self.universe.private_files and oct(S_IMODE(stat(self.filename)[ST_MODE])) != 0600:
+ chmod(self.filename, 0600)
+
+ # write it back sorted, instead of using ConfigParser
+ sections = self.data.sections()
+ sections.sort()
+ for section in sections:
+ file_descriptor.write("[" + section + "]\n")
+ options = self.data.options(section)
+ options.sort()
+ for option in options:
+ file_descriptor.write(option + " = " + self.data.get(section, option) + "\n")
+ file_descriptor.write("\n")
+
+ # flush and close the file
file_descriptor.flush()
file_descriptor.close()
+ # unset the modified flag
+ self.modified = False
+ def is_writeable(self):
+ """Returns True if the __control__ read_only is False."""
+ return not self.data.has_option("__control__", "read_only") or not self.data.getboolean("__control__", "read_only")
+
class Universe:
"""The universe."""
+
def __init__(self, filename=""):
"""Initialize the universe."""
self.categories = {}
self.contents = {}
self.default_origins = {}
- self.files = {}
+ self.private_files = []
+ self.loglines = []
+ self.pending_events_long = {}
+ self.pending_events_short = {}
self.userlist = []
self.terminate_world = False
self.reload_modules = False
]
for filename in possible_filenames:
if access(filename, R_OK): break
- if not filename.startswith(sep):
- filename = getcwd() + sep + filename
- DataFile(filename, self)
+ if not isabs(filename):
+ filename = abspath(filename)
+ self.filename = filename
+ self.load()
+
+ def load(self):
+ """Load universe data from persistent storage."""
+
+ # the files dict must exist and filename needs to be read-only
+ if not hasattr(self, "files") or not ( self.filename in self.files and self.files[self.filename].is_writeable() ):
+
+ # clear out all read-only files
+ if hasattr(self, "files"):
+ for data_filename in self.files.keys():
+ if not self.files[data_filename].is_writeable():
+ del self.files[data_filename]
+
+ # start loading from the initial file
+ DataFile(self.filename, self)
+
+ # make a list of inactive avatars
+ inactive_avatars = []
+ for account in self.categories["account"].values():
+ inactive_avatars += [ (self.contents[x]) for x in account.getlist("avatars") ]
+ for user in self.userlist:
+ if user.avatar in inactive_avatars:
+ inactive_avatars.remove(user.avatar)
+
+ # go through all elements to clear out inactive avatar locations
+ for element in self.contents.values():
+ location = element.get("location")
+ if element in inactive_avatars and location:
+ if location in self.contents and element.key in self.contents[location].contents:
+ del self.contents[location].contents[element.key]
+ element.set("default_location", location)
+ element.remove_facet("location")
+
+ # another pass to straighten out all the element contents
+ for element in self.contents.values():
+ element.update_location()
+ element.clean_contents()
+
def save(self):
"""Save the universe to persistent storage."""
for key in self.files: self.files[key].save()
self.listening_socket.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
# bind the socket to to our desired server ipa and port
- self.listening_socket.bind((self.categories["internal"]["network"].get("host"), self.categories["internal"]["network"].getint("port")))
+ host = self.categories["internal"]["network"].get("host")
+ port = self.categories["internal"]["network"].getint("port")
+ self.listening_socket.bind((host, port))
# disable blocking so we can proceed whether or not we can
# send/receive
self.listening_socket.listen(1)
# note that we're now ready for user connections
- log("Waiting for connection(s)...")
+ if not host: host = "0.0.0.0"
+ log("Listening for Telnet connections on: " + host + ":" + str(port))
+
+ def get_time(self):
+ """Convenience method to get the elapsed time counter."""
+ return self.categories["internal"]["counters"].getint("elapsed")
class User:
"""This is a connected user."""
def __init__(self):
"""Default values for the in-memory user variables."""
+ self.account = None
self.address = ""
- self.last_address = ""
- self.connection = None
self.authenticated = False
- self.password_tries = 1
- self.state = "entering_account_name"
- self.menu_seen = False
+ self.avatar = None
+ self.connection = None
+ self.echoing = True
self.error = ""
self.input_queue = []
+ self.last_address = ""
+ self.menu_choices = {}
+ self.menu_seen = False
+ self.negotiation_pause = 0
self.output_queue = []
self.partial_input = ""
- self.echoing = True
- self.avatar = None
- self.account = None
+ self.password_tries = 0
+ self.received_newline = True
+ self.state = "initial"
+ self.terminator = IAC+GA
def quit(self):
"""Log, close the connection and remove."""
- name = self.account.get("name")
+ if self.account: name = self.account.get("name")
+ else: name = ""
if name: message = "User " + name
else: message = "An unnamed user"
message += " logged out."
- log(message)
+ log(message, 2)
+ self.deactivate_avatar()
self.connection.close()
self.remove()
# create a new user object
new_user = User()
- # set everything else equivalent
- for attribute in [
- "address",
- "last_address",
- "connection",
- "authenticated",
- "password_tries",
- "state",
- "menu_seen",
- "error",
- "input_queue",
- "output_queue",
- "partial_input",
- "echoing",
- "avatar",
- "account"
- ]:
+ # set everything equivalent
+ for attribute in vars(self).keys():
exec("new_user." + attribute + " = self." + attribute)
+ # the avatar needs a new owner
+ if new_user.avatar: new_user.avatar.owner = new_user
+
# add it to the list
universe.userlist.append(new_user)
if old_user.account.get("name") == self.account.get("name") and old_user is not self:
# make a note of it
- log("User " + self.account.get("name") + " reconnected--closing old connection to " + old_user.address + ".")
- old_user.send("$(eol)$(red)New connection from " + self.address + ". Terminating old connection...$(nrm)$(eol)")
- self.send("$(eol)$(red)Taking over old connection from " + old_user.address + ".$(nrm)")
+ log("User " + self.account.get("name") + " reconnected--closing old connection to " + old_user.address + ".", 2)
+ old_user.send("$(eol)$(red)New connection from " + self.address + ". Terminating old connection...$(nrm)$(eol)", flush=True, add_prompt=False)
# close the old connection
old_user.connection.close()
# replace the old connection with this one
+ old_user.send("$(eol)$(red)Taking over old connection from " + old_user.address + ".$(nrm)")
old_user.connection = self.connection
old_user.last_address = old_user.address
old_user.address = self.address
- old_user.echoing = self.echoing
+
+ # may need to tell the new connection to echo
+ if old_user.echoing:
+ old_user.send(get_echo_sequence(old_user.state, self.echoing), raw=True)
# take this one out of the list and delete
self.remove()
def authenticate(self):
"""Flag the user as authenticated and disconnect duplicates."""
if not self.state is "authenticated":
- log("User " + self.account.get("name") + " logged in.")
+ log("User " + self.account.get("name") + " logged in.", 2)
self.authenticated = True
+ if self.account.subkey in universe.categories["internal"]["limits"].getlist("default_admins"):
+ self.account.set("administrator", "True")
def show_menu(self):
"""Send the user their current menu."""
if not self.menu_seen:
self.menu_choices = get_menu_choices(self)
- self.send(get_menu(self.state, self.error, self.echoing, self.menu_choices), "")
+ self.send(get_menu(self.state, self.error, self.echoing, self.terminator, self.menu_choices), "")
self.menu_seen = True
self.error = False
self.adjust_echoing()
"""Remove a user from the list of connected users."""
universe.userlist.remove(self)
- def send(self, output, eol="$(eol)"):
+ def send(self, output, eol="$(eol)", raw=False, flush=False, add_prompt=True, just_prompt=False):
"""Send arbitrary text to a connected user."""
- # only when there is actual output
- #if output:
+ # unless raw mode is on, clean it up all nice and pretty
+ if not raw:
- # start with a newline, append the message, then end
- # with the optional eol string passed to this function
- # and the ansi escape to return to normal text
- output = "\r\n" + output + eol + chr(27) + "[0m"
+ # strip extra $(eol) off if present
+ while output.startswith("$(eol)"): output = output[6:]
+ while output.endswith("$(eol)"): output = output[:-6]
- # find and replace macros in the output
- output = replace_macros(self, output)
+ # we'll take out GA or EOR and add them back on the end
+ if output.endswith(IAC+GA) or output.endswith(IAC+EOR):
+ terminate = True
+ output = output[:-2]
+ else: terminate = False
- # wrap the text at 80 characters
- # TODO: prompt user for preferred wrap width
- output = wrap_ansi_text(output, 80)
+ # start with a newline, append the message, then end
+ # with the optional eol string passed to this function
+ # and the ansi escape to return to normal text
+ if not just_prompt: output = "$(eol)$(eol)" + output
+ output += eol + chr(27) + "[0m"
- # drop the formatted output into the output queue
- self.output_queue.append(output)
+ # tack on a prompt if active
+ if self.state == "active":
+ if not just_prompt: output += "$(eol)"
+ if add_prompt: output += "> "
- # try to send the last item in the queue, remove it and
- # flag that menu display is not needed
- try:
- self.connection.send(self.output_queue[0])
- self.output_queue.remove(self.output_queue[0])
- self.menu_seen = False
+ # find and replace macros in the output
+ output = replace_macros(self, output)
- # but if we can't, that's okay too
- except:
- pass
+ # wrap the text at 80 characters
+ output = wrap_ansi_text(output, 80)
+
+ # tack the terminator back on
+ if terminate: output += self.terminator
+
+ # drop the output into the user's output queue
+ self.output_queue.append(output)
+
+ # if this is urgent, flush all pending output
+ if flush: self.flush()
def pulse(self):
"""All the things to do to the user per increment."""
self.state = "disconnecting"
self.menu_seen = False
+ # if output is paused, decrement the counter
+ if self.state == "initial":
+ if self.negotiation_pause: self.negotiation_pause -= 1
+ else: self.state = "entering_account_name"
+
# show the user a menu as needed
- self.show_menu()
+ elif not self.state == "active": self.show_menu()
+
+ # flush any pending output in teh queue
+ self.flush()
# disconnect users with the appropriate state
- if self.state == "disconnecting":
- self.quit()
+ if self.state == "disconnecting": self.quit()
- # the user is unique and not flagged to disconnect
- else:
-
- # check for input and add it to the queue
- self.enqueue_input()
+ # check for input and add it to the queue
+ self.enqueue_input()
+
+ # there is input waiting in the queue
+ if self.input_queue: handle_user_input(self)
+
+ def flush(self):
+ """Try to send the last item in the queue and remove it."""
+ if self.output_queue:
+ if self.received_newline:
+ self.received_newline = False
+ if self.output_queue[0].startswith("\r\n"):
+ self.output_queue[0] = self.output_queue[0][2:]
+ try:
+ self.connection.send(self.output_queue[0])
+ del self.output_queue[0]
+ except:
+ pass
- # there is input waiting in the queue
- if self.input_queue: handle_user_input(self)
def enqueue_input(self):
"""Process and enqueue any new input."""
# tack this on to any previous partial
self.partial_input += input_data
+ # reply to and remove any IAC negotiation codes
+ self.negotiate_telnet_options()
+
# separate multiple input lines
new_input_lines = self.partial_input.split("\n")
# iterate over the remaining lines
for line in new_input_lines:
+ # remove a trailing carriage return
+ if line.endswith("\r"): line = line.rstrip("\r")
+
+ # log non-printable characters remaining
+ removed = filter(lambda x: (x < " " or x > "~"), line)
+ if removed:
+ logline = "Non-printable characters from "
+ if self.account and self.account.get("name"): logline += self.account.get("name") + ": "
+ else: logline += "unknown user: "
+ logline += repr(removed)
+ log(logline, 4)
+
# filter out non-printables
- line = filter(lambda x: x>=' ' and x<='~', line)
+ line = filter(lambda x: " " <= x <= "~", line)
# strip off extra whitespace
line = line.strip()
# put on the end of the queue
self.input_queue.append(line)
+ def negotiate_telnet_options(self):
+ """Reply to/remove partial_input telnet negotiation options."""
+
+ # start at the begining of the input
+ position = 0
+
+ # make a local copy to play with
+ text = self.partial_input
+
+ # as long as we haven't checked it all
+ while position < len(text):
+
+ # jump to the first IAC you find
+ position = text.find(IAC, position)
+
+ # if there wasn't an IAC in the input, skip to the end
+ if position < 0: position = len(text)
+
+ # replace a double (literal) IAC if there's an LF later
+ elif len(text) > position+1 and text[position+1] == IAC:
+ if text.find("\n", position) > 0: text = text.replace(IAC+IAC, IAC)
+ else: position += 1
+ position += 1
+
+ # this must be an option negotiation
+ elif len(text) > position+2 and text[position+1] in (DO, DONT, WILL, WONT):
+
+ negotiation = text[position+1:position+3]
+
+ # if we turned echo off, ignore the confirmation
+ if not self.echoing and negotiation == DO+ECHO: pass
+
+ # allow LINEMODE
+ elif negotiation == WILL+LINEMODE: self.send(IAC+DO+LINEMODE, raw=True)
+
+ # if the client likes EOR instead of GA, make a note of it
+ elif negotiation == DO+EOR: self.terminator = IAC+EOR
+ elif negotiation == DONT+EOR and self.terminator == IAC+EOR:
+ self.terminator = IAC+GA
+
+ # if the client doesn't want GA, oblige
+ elif negotiation == DO+SGA and self.terminator == IAC+GA:
+ self.terminator = ""
+ self.send(IAC+WILL+SGA, raw=True)
+
+ # we don't want to allow anything else
+ elif text[position+1] == DO: self.send(IAC+WONT+text[position+2], raw=True)
+ elif text[position+1] == WILL: self.send(IAC+DONT+text[position+2], raw=True)
+
+ # strip the negotiation from the input
+ text = text.replace(text[position:position+3], "")
+
+ # get rid of IAC SB .* IAC SE
+ elif len(text) > position+4 and text[position:position+2] == IAC+SB:
+ end_subnegotiation = text.find(IAC+SE, position)
+ if end_subnegotiation > 0: text = text[:position] + text[end_subnegotiation+2:]
+ else: position += 1
+
+ # otherwise, strip out a two-byte IAC command
+ elif len(text) > position+2: text = text.replace(text[position:position+2], "")
+
+ # and this means we got the begining of an IAC
+ else: position += 1
+
+ # replace the input with our cleaned-up text
+ self.partial_input = text
+
def new_avatar(self):
"""Instantiate a new, unconfigured avatar for this user."""
- counter = universe.categories["internal"]["counters"].getint("next_avatar")
- while "avatar:" + repr(counter + 1) in universe.categories["actor"].keys(): counter += 1
- universe.categories["internal"]["counters"].set("next_avatar", counter + 1)
- self.avatar = Element("actor:avatar:" + repr(counter), universe)
- avatars = self.account.get("avatars").split()
- avatars.append(self.avatar.key)
- self.account.set("avatars", " ".join(avatars))
+ counter = 0
+ while "avatar:" + self.account.get("name") + ":" + str(counter) in universe.categories["actor"].keys(): counter += 1
+ self.avatar = Element("actor:avatar:" + self.account.get("name") + ":" + str(counter), universe)
+ self.avatar.append("inherit", "archetype:avatar")
+ self.account.append("avatars", self.avatar.key)
+
+ def delete_avatar(self, avatar):
+ """Remove an avatar from the world and from the user's list."""
+ if self.avatar is universe.contents[avatar]: self.avatar = None
+ universe.contents[avatar].destroy()
+ avatars = self.account.getlist("avatars")
+ avatars.remove(avatar)
+ self.account.set("avatars", avatars)
+
+ def activate_avatar_by_index(self, index):
+ """Enter the world with a particular indexed avatar."""
+ self.avatar = universe.contents[self.account.getlist("avatars")[index]]
+ self.avatar.owner = self
+ self.state = "active"
+ self.avatar.go_home()
+
+ def deactivate_avatar(self):
+ """Have the active avatar leave the world."""
+ if self.avatar:
+ current = self.avatar.get("location")
+ self.avatar.set("default_location", current)
+ self.avatar.echo_to_location("You suddenly wonder where " + self.avatar.get("name") + " went.")
+ del universe.contents[current].contents[self.avatar.key]
+ self.avatar.remove_facet("location")
+ self.avatar.owner = None
+ self.avatar = None
+
+ def destroy(self):
+ """Destroy the user and associated avatars."""
+ for avatar in self.account.getlist("avatars"): self.delete_avatar(avatar)
+ self.account.destroy()
def list_avatar_names(self):
- """A test function to list names of assigned avatars."""
- try:
- avatars = self.account.get("avatars").split()
- except:
- avatars = []
- avatar_names = []
- for avatar in avatars:
- avatar_names.append(universe.contents[avatar].get("name"))
- return avatar_names
-
-def broadcast(message):
+ """List names of assigned avatars."""
+ return [ universe.contents[avatar].get("name") for avatar in self.account.getlist("avatars") ]
+
+def create_pidfile(universe):
+ """Write a file containing the current process ID."""
+ pid = str(getpid())
+ log("Process ID: " + pid)
+ file_name = universe.contents["internal:process"].get("pidfile")
+ if file_name:
+ file_descriptor = file(file_name, 'w')
+ file_descriptor.write(pid + "\n")
+ file_descriptor.flush()
+ file_descriptor.close()
+
+def remove_pidfile(universe):
+ """Remove the file containing the current process ID."""
+ file_name = universe.contents["internal:process"].get("pidfile")
+ if file_name and access(file_name, W_OK): remove(file_name)
+
+def makelist(value):
+ """Turn string into list type."""
+ if value[0] + value[-1] == "[]": return eval(value)
+ else: return [ value ]
+
+def makedict(value):
+ """Turn string into dict type."""
+ if value[0] + value[-1] == "{}": return eval(value)
+ elif value.find(":") > 0: return eval("{" + value + "}")
+ else: return { value: None }
+
+def broadcast(message, add_prompt=True):
"""Send a message to all connected users."""
- for each_user in universe.userlist: each_user.send("$(eol)" + message)
+ for each_user in universe.userlist: each_user.send("$(eol)" + message, add_prompt=add_prompt)
-def log(message):
+def log(message, level=0):
"""Log a message."""
- # the time in posix log timestamp format
+ # a couple references we need
+ file_name = universe.categories["internal"]["logging"].get("file")
+ max_log_lines = universe.categories["internal"]["logging"].getint("max_log_lines")
+ syslog_name = universe.categories["internal"]["logging"].get("syslog")
timestamp = asctime()[4:19]
- # send the timestamp and message to standard output
- print(timestamp + " " + message)
+ # turn the message into a list of lines
+ lines = filter(lambda x: x!="", [(x.rstrip()) for x in message.split("\n")])
+
+ # send the timestamp and line to a file
+ if file_name:
+ file_descriptor = file(file_name, "a")
+ for line in lines: file_descriptor.write(timestamp + " " + line + "\n")
+ file_descriptor.flush()
+ file_descriptor.close()
+
+ # send the timestamp and line to standard output
+ if universe.categories["internal"]["logging"].getboolean("stdout"):
+ for line in lines: print(timestamp + " " + line)
+
+ # send the line to the system log
+ if syslog_name:
+ openlog(syslog_name, LOG_PID, LOG_INFO | LOG_DAEMON)
+ for line in lines: syslog(line)
+ closelog()
+
+ # display to connected administrators
+ for user in universe.userlist:
+ if user.state == "active" and user.account.getboolean("administrator") and user.account.getint("loglevel") <= level:
+ # iterate over every line in the message
+ full_message = ""
+ for line in lines:
+ full_message += "$(bld)$(red)" + timestamp + " " + line + "$(nrm)$(eol)"
+ user.send(full_message, flush=True)
+
+ # add to the recent log list
+ for line in lines:
+ while 0 < len(universe.loglines) >= max_log_lines: del universe.loglines[0]
+ universe.loglines.append((level, timestamp + " " + line))
+
+def get_loglines(level, start, stop):
+ """Return a specific range of loglines filtered by level."""
+
+ # filter the log lines
+ loglines = filter(lambda x: x[0]>=level, universe.loglines)
+
+ # we need these in several places
+ total_count = str(len(universe.loglines))
+ filtered_count = len(loglines)
+
+ # don't proceed if there are no lines
+ if filtered_count:
+
+ # can't start before the begining or at the end
+ if start > filtered_count: start = filtered_count
+ if start < 1: start = 1
+
+ # can't stop before we start
+ if stop > start: stop = start
+ elif stop < 1: stop = 1
+
+ # some preamble
+ message = "There are " + str(total_count)
+ message += " log lines in memory and " + str(filtered_count)
+ message += " at or above level " + str(level) + "."
+ message += " The lines from " + str(stop) + " to " + str(start)
+ message += " are:$(eol)$(eol)"
+
+ # add the text from the selected lines
+ if stop > 1: range_lines = loglines[-start:-(stop-1)]
+ else: range_lines = loglines[-start:]
+ for line in range_lines:
+ message += " (" + str(line[0]) + ") " + line[1] + "$(eol)"
+
+ # there were no lines
+ else:
+ message = "None of the " + str(total_count)
+ message += " lines in memory matches your request."
+
+ # pass it back
+ return message
def wrap_ansi_text(text, width):
"""Wrap text with arbitrary width while ignoring ANSI colors."""
def replace_macros(user, text, is_input=False):
"""Replaces macros in text output."""
+
+ # loop until broken
while True:
+
+ # third person pronouns
+ pronouns = {
+ "female": { "obj": "her", "pos": "hers", "sub": "she" },
+ "male": { "obj": "him", "pos": "his", "sub": "he" },
+ "neuter": { "obj": "it", "pos": "its", "sub": "it" }
+ }
+
+ # a dict of replacement macros
+ macros = {
+ "$(eol)": "\r\n",
+ "$(bld)": chr(27) + "[1m",
+ "$(nrm)": chr(27) + "[0m",
+ "$(blk)": chr(27) + "[30m",
+ "$(blu)": chr(27) + "[34m",
+ "$(cyn)": chr(27) + "[36m",
+ "$(grn)": chr(27) + "[32m",
+ "$(mgt)": chr(27) + "[35m",
+ "$(red)": chr(27) + "[31m",
+ "$(yel)": chr(27) + "[33m",
+ }
+
+ # add dynamic macros where possible
+ if user.account:
+ account_name = user.account.get("name")
+ if account_name:
+ macros["$(account)"] = account_name
+ if user.avatar:
+ avatar_gender = user.avatar.get("gender")
+ if avatar_gender:
+ macros["$(tpop)"] = pronouns[avatar_gender]["obj"]
+ macros["$(tppp)"] = pronouns[avatar_gender]["pos"]
+ macros["$(tpsp)"] = pronouns[avatar_gender]["sub"]
+
+ # find and replace per the macros dict
macro_start = text.find("$(")
if macro_start == -1: break
macro_end = text.find(")", macro_start) + 1
if macro in macros.keys():
text = text.replace(macro, macros[macro])
- # the user's account name
- elif macro == "$(account)":
- text = text.replace(macro, user.account.get("name"))
-
- # third person subjective pronoun
- elif macro == "$(tpsp)":
- if user.avatar.get("gender") == "male":
- text = text.replace(macro, "he")
- elif user.avatar.get("gender") == "female":
- text = text.replace(macro, "she")
- else:
- text = text.replace(macro, "it")
-
- # third person objective pronoun
- elif macro == "$(tpop)":
- if user.avatar.get("gender") == "male":
- text = text.replace(macro, "him")
- elif user.avatar.get("gender") == "female":
- text = text.replace(macro, "her")
- else:
- text = text.replace(macro, "it")
-
- # third person possessive pronoun
- elif macro == "$(tppp)":
- if user.avatar.get("gender") == "male":
- text = text.replace(macro, "his")
- elif user.avatar.get("gender") == "female":
- text = text.replace(macro, "hers")
- else:
- text = text.replace(macro, "its")
-
# if we get here, log and replace it with null
else:
text = text.replace(macro, "")
if not is_input:
- log("Unexpected replacement macro " + macro + " encountered.")
+ log("Unexpected replacement macro " + macro + " encountered.", 6)
# replace the look-like-a-macro sequence
text = text.replace("$_(", "$(")
return text
+def escape_macros(text):
+ """Escapes replacement macros in text."""
+ return text.replace("$(", "$_(")
+
def check_time(frequency):
"""Check for a factor of the current increment count."""
if type(frequency) is str:
# update the log every now and then
if check_time("frequency_log"):
- log(repr(len(universe.userlist)) + " connection(s)")
+ log(str(len(universe.userlist)) + " connection(s)")
# periodically save everything
if check_time("frequency_save"):
# pause for a configurable amount of time (decimal seconds)
sleep(universe.categories["internal"]["time"].getfloat("increment"))
- # increment the elapsed increment counter
+ # increase the elapsed increment counter
universe.categories["internal"]["counters"].set("elapsed", universe.categories["internal"]["counters"].getint("elapsed") + 1)
def reload_data():
- """Reload data into new persistent objects."""
+ """Reload all relevant objects."""
for user in universe.userlist[:]: user.reload()
+ for element in universe.contents.values():
+ if element.origin.is_writeable(): element.reload()
+ universe.load()
def check_for_connection(listening_socket):
"""Check for a waiting connection and return a new user object."""
return None
# note that we got one
- log("Connection from " + address[0])
+ log("Connection from " + address[0], 2)
# disable blocking so we can proceed whether or not we can send/receive
connection.setblocking(0)
# set the user's ipa from the connection's ipa
user.address = address[0]
+ # let the client know we WILL EOR
+ user.send(IAC+WILL+EOR, raw=True)
+ user.negotiation_pause = 2
+
# return the new user object
return user
-def get_menu(state, error=None, echoing=True, choices={}):
+def get_menu(state, error=None, echoing=True, terminator="", choices=None):
"""Show the correct menu text to a user."""
+ # make sure we don't reuse a mutable sequence by default
+ if choices is None: choices = {}
+
# begin with a telnet echo command sequence if needed
message = get_echo_sequence(state, echoing)
# display a message indicating if echo is off
message += get_echo_message(state)
+ # tack on EOR or GA to indicate the prompt will not be followed by CRLF
+ message += terminator
+
# return the assembly of various strings defined above
return message
return universe.categories["menu"][state].getboolean("echo", True)
def get_echo_sequence(state, echoing):
- """Build the appropriate IAC ECHO sequence as needed."""
+ """Build the appropriate IAC WILL/WONT ECHO sequence as needed."""
# if the user has echo on and the menu specifies it should be turned
# off, send: iac + will + echo + null
- if echoing and not menu_echo_on(state): return chr(255) + chr(251) + chr(1) + chr(0)
+ if echoing and not menu_echo_on(state): return IAC+WILL+ECHO
# if echo is not set to off in the menu and the user curently has echo
# off, send: iac + wont + echo + null
- elif not echoing and menu_echo_on(state): return chr(255) + chr(252) + chr(1) + chr(0)
+ elif not echoing and menu_echo_on(state): return IAC+WONT+ECHO
# default is not to send an echo control sequence at all
else: return ""
def get_formatted_default_menu_choice(state):
"""Default menu choice foratted for inclusion in a prompt string."""
- default = get_default_menu_choice(state)
- if default: return "[$(red)" + default + "$(nrm)] "
+ default_choice = get_default_menu_choice(state)
+ if default_choice: return "[$(red)" + default_choice + "$(nrm)] "
else: return ""
def get_menu_description(state, error):
def get_menu_choices(user):
"""Return a dict of choice:meaning."""
- choices = {}
- for facet in universe.categories["menu"][user.state].facets():
- if facet.startswith("choice_"):
- choices[facet.split("_", 2)[1]] = universe.categories["menu"][user.state].get(facet)
+ menu = universe.categories["menu"][user.state]
+ create_choices = menu.get("create")
+ if create_choices: choices = eval(create_choices)
+ else: choices = {}
+ ignores = []
+ options = {}
+ creates = {}
+ for facet in menu.facets():
+ if facet.startswith("demand_") and not eval(universe.categories["menu"][user.state].get(facet)):
+ ignores.append(facet.split("_", 2)[1])
elif facet.startswith("create_"):
- choices[facet.split("_", 2)[1]] = eval(universe.categories["menu"][user.state].get(facet))
+ creates[facet] = facet.split("_", 2)[1]
+ elif facet.startswith("choice_"):
+ options[facet] = facet.split("_", 2)[1]
+ for facet in creates.keys():
+ if not creates[facet] in ignores:
+ choices[creates[facet]] = eval(menu.get(facet))
+ for facet in options.keys():
+ if not options[facet] in ignores:
+ choices[options[facet]] = menu.get(facet)
return choices
def get_formatted_menu_choices(state, choices):
def get_choice_branch(user, choice):
"""Returns the new state matching the given choice."""
branches = get_menu_branches(user.state)
- if not choice: choice = get_default_menu_choice(user.state)
if choice in branches.keys(): return branches[choice]
elif choice in user.menu_choices.keys(): return get_default_branch(user.state)
else: return ""
def get_choice_action(user, choice):
"""Run any indicated script for the given choice."""
actions = get_menu_actions(user.state)
- if not choice: choice = get_default_menu_choice(user.state)
if choice in actions.keys(): return actions[choice]
elif choice in user.menu_choices.keys(): return get_default_action(user.state)
else: return ""
user.menu_seen = False
# if the user's client echo is off, send a blank line for aesthetics
- if not user.echoing: user.send("", "")
+ if user.echoing: user.received_newline = True
def generic_menu_handler(user):
"""A generic menu choice handler."""
choice = user.input_queue.pop(0)
if choice: choice = choice.lower()
else: choice = ""
-
- # run any script related to this choice
- exec(get_choice_action(user, choice))
-
- # move on to the next state or return an error
- new_state = get_choice_branch(user, choice)
- if new_state: user.state = new_state
+ if not choice: choice = get_default_menu_choice(user.state)
+ if choice in user.menu_choices:
+ exec(get_choice_action(user, choice))
+ new_state = get_choice_branch(user, choice)
+ if new_state: user.state = new_state
else: user.error = "default"
def handler_entering_account_name(user):
else:
user.account = Element("account:" + name, universe)
user.account.set("name", name)
- log("New user: " + name)
+ log("New user: " + name, 2)
user.state = "checking_new_account_name"
# if the user entered nothing for a name, then buhbye
user.state = "main_utility"
# if at first your hashes don't match, try, try again
- elif user.password_tries < universe.categories["internal"]["general"].getint("password_tries"):
+ elif user.password_tries < universe.categories["internal"]["limits"].getint("password_tries") - 1:
user.password_tries += 1
user.error = "incorrect"
user.send("$(eol)$(red)Too many failed password attempts...$(nrm)$(eol)")
user.state = "disconnecting"
-def handler_checking_new_account_name(user):
- """Handle input for the new user menu."""
-
- # get the next waiting line of input
- input_data = user.input_queue.pop(0)
-
- # if there's input, take the first character and lowercase it
- if input_data:
- choice = input_data.lower()[0]
-
- # if there's no input, use the default
- else:
- choice = get_default_menu_choice(user.state)
-
- # user selected to disconnect
- if choice == "d":
- user.account.delete()
- user.state = "disconnecting"
-
- # go back to the login screen
- elif choice == "g":
- user.account.delete()
- user.state = "entering_account_name"
-
- # new user, so ask for a password
- elif choice == "n":
- user.state = "entering_new_password"
-
- # user entered a non-existent option
- else:
- user.error = "default"
-
def handler_entering_new_password(user):
"""Handle a new password entry."""
user.state = "verifying_new_password"
# the password was weak, try again if you haven't tried too many times
- elif user.password_tries < universe.categories["internal"]["general"].getint("password_tries"):
+ elif user.password_tries < universe.categories["internal"]["limits"].getint("password_tries") - 1:
user.password_tries += 1
user.error = "weak"
# too many tries, so adios
else:
user.send("$(eol)$(red)Too many failed password attempts...$(nrm)$(eol)")
- user.account.delete()
+ user.account.destroy()
user.state = "disconnecting"
def handler_verifying_new_password(user):
# go back to entering the new password as long as you haven't tried
# too many times
- elif user.password_tries < universe.categories["internal"]["general"].getint("password_tries"):
+ elif user.password_tries < universe.categories["internal"]["limits"].getint("password_tries") - 1:
user.password_tries += 1
user.error = "differs"
user.state = "entering_new_password"
# otherwise, sayonara
else:
user.send("$(eol)$(red)Too many failed password attempts...$(nrm)$(eol)")
- user.account.delete()
+ user.account.destroy()
user.state = "disconnecting"
def handler_active(user):
# get the next waiting line of input
input_data = user.input_queue.pop(0)
- # split out the command (first word) and parameters (everything else)
- if input_data.find(" ") > 0:
- command, parameters = input_data.split(" ", 1)
- else:
- command = input_data
- parameters = ""
+ # is there input?
+ if input_data:
- # lowercase the command
- command = command.lower()
+ # split out the command (first word) and parameters (everything else)
+ if input_data.find(" ") > 0:
+ command_name, parameters = input_data.split(" ", 1)
+ else:
+ command_name = input_data
+ parameters = ""
+
+ # lowercase the command
+ command_name = command_name.lower()
- # the command matches a command word for which we have data
- if command in universe.categories["command"]:
- exec(universe.categories["command"][command].get("action"))
+ # the command matches a command word for which we have data
+ if command_name in universe.categories["command"]:
+ command = universe.categories["command"][command_name]
+ else: command = None
- # no data matching the entered command word
- elif command: command_error(user, command, parameters)
+ # if it's allowed, do it
+ actor = user.avatar
+ if actor.can_run(command): exec(command.get("action"))
-def command_halt(user, command="", parameters=""):
+ # otherwise, give an error
+ elif command_name: command_error(actor, input_data)
+
+ # if no input, just idle back with a prompt
+ else: user.send("", just_prompt=True)
+
+def command_halt(actor, parameters):
"""Halt the world."""
+ if actor.owner:
- # see if there's a message or use a generic one
- if parameters: message = "Halting: " + parameters
- else: message = "User " + user.account.get("name") + " halted the world."
+ # see if there's a message or use a generic one
+ if parameters: message = "Halting: " + parameters
+ else: message = "User " + actor.owner.account.get("name") + " halted the world."
- # let everyone know
- broadcast(message)
- log(message)
+ # let everyone know
+ broadcast(message, add_prompt=False)
+ log(message, 8)
- # set a flag to terminate the world
- universe.terminate_world = True
+ # set a flag to terminate the world
+ universe.terminate_world = True
-def command_reload(user, command="", parameters=""):
+def command_reload(actor):
"""Reload all code modules, configs and data."""
+ if actor.owner:
- # let the user know and log
- user.send("Reloading all code modules, configs and data.")
- log("User " + user.account.get("name") + " reloaded the world.")
+ # let the user know and log
+ actor.send("Reloading all code modules, configs and data.")
+ log("User " + actor.owner.account.get("name") + " reloaded the world.", 8)
- # set a flag to reload
- universe.reload_modules = True
+ # set a flag to reload
+ universe.reload_modules = True
-def command_quit(user, command="", parameters=""):
- """Quit the world."""
- user.state = "disconnecting"
+def command_quit(actor):
+ """Leave the world and go back to the main menu."""
+ if actor.owner:
+ actor.owner.state = "main_utility"
+ actor.owner.deactivate_avatar()
-def command_help(user, command="", parameters=""):
+def command_help(actor, parameters):
"""List available commands and provide help for commands."""
# did the user ask for help on a specific command word?
- if parameters:
+ if parameters and actor.owner:
# is the command word one for which we have data?
if parameters in universe.categories["command"]:
+ command = universe.categories["command"][parameters]
+ else: command = None
+
+ # only for allowed commands
+ if actor.can_run(command):
# add a description if provided
- description = universe.categories["command"][parameters].get("description")
+ description = command.get("description")
if not description:
description = "(no short description provided)"
- output = "$(grn)" + parameters + "$(nrm) - " + description + "$(eol)$(eol)"
+ if command.getboolean("administrative"): output = "$(red)"
+ else: output = "$(grn)"
+ output += parameters + "$(nrm) - " + description + "$(eol)$(eol)"
# add the help text if provided
- help_text = universe.categories["command"][parameters].get("help")
+ help_text = command.get("help")
if not help_text:
help_text = "No help is provided for this command."
output += help_text
sorted_commands = universe.categories["command"].keys()
sorted_commands.sort()
for item in sorted_commands:
- description = universe.categories["command"][item].get("description")
- if not description:
- description = "(no short description provided)"
- output += " $(grn)" + item + "$(nrm) - " + description + "$(eol)"
+ command = universe.categories["command"][item]
+ if actor.can_run(command):
+ description = command.get("description")
+ if not description:
+ description = "(no short description provided)"
+ if command.getboolean("administrative"): output += " $(red)"
+ else: output += " $(grn)"
+ output += item + "$(nrm) - " + description + "$(eol)"
output += "$(eol)Enter \"help COMMAND\" for help on a command named \"COMMAND\"."
# send the accumulated output to the user
- user.send(output)
+ actor.send(output)
+
+def command_move(actor, parameters):
+ """Move the avatar in a given direction."""
+ if parameters in universe.contents[actor.get("location")].portals():
+ actor.move_direction(parameters)
+ else: actor.send("You cannot go that way.")
-def command_say(user, command="", parameters=""):
+def command_look(actor, parameters):
+ """Look around."""
+ if parameters: actor.send("You can't look at or in anything yet.")
+ else: actor.look_at(actor.get("location"))
+
+def command_say(actor, parameters):
"""Speak to others in the same room."""
# check for replacement macros
- if replace_macros(user, parameters, True) != parameters:
- user.send("You cannot speak $_(replacement macros).")
+ if replace_macros(actor.owner, parameters, True) != parameters:
+ actor.send("You cannot speak $_(replacement macros).")
# the user entered a message
elif parameters:
# get rid of quote marks on the ends of the message and
# capitalize the first letter
- message = parameters.strip("\"'`").capitalize()
+ message = parameters.strip("\"'`")
+ message = message[0].capitalize() + message[1:]
# a dictionary of punctuation:action pairs
actions = {}
for facet in universe.categories["internal"]["language"].facets():
if facet.startswith("punctuation_"):
action = facet.split("_")[1]
- for mark in universe.categories["internal"]["language"].get(facet).split():
+ for mark in universe.categories["internal"]["language"].getlist(facet):
actions[mark] = action
# match the punctuation used, if any, to an action
message += default_punctuation
# capitalize a list of words within the message
- capitalize = universe.categories["internal"]["language"].get("capitalize").split()
- for word in capitalize:
+ capitalize_words = universe.categories["internal"]["language"].getlist("capitalize_words")
+ for word in capitalize_words:
message = message.replace(" " + word + " ", " " + word.capitalize() + " ")
# tell the room
- # TODO: we won't be using broadcast once there are actual rooms
- broadcast(user.account.get("name") + " " + action + "s, \"" + message + "\"")
+ actor.echo_to_location(actor.get("name") + " " + action + "s, \"" + message + "\"")
+ actor.send("You " + action + ", \"" + message + "\"")
# there was no message
else:
- user.send("What do you want to say?")
+ actor.send("What do you want to say?")
-def command_show(user, command="", parameters=""):
+def command_show(actor, parameters):
"""Show program data."""
- if parameters == "avatars":
- message = "These are the avatars managed by your account:$(eol)"
- avatars = user.list_avatar_names()
- avatars.sort()
- for avatar in avatars: message += "$(eol) $(grn)" + avatar + "$(nrm)"
- elif parameters == "files":
- message = "These are the current files containing the universe:$(eol)"
- keys = universe.files.keys()
- keys.sort()
- for key in keys: message += "$(eol) $(grn)" + key + "$(nrm)"
- elif parameters == "universe":
- message = "These are the current elements in the universe:$(eol)"
- keys = universe.contents.keys()
- keys.sort()
- for key in keys: message += "$(eol) $(grn)" + key + "$(nrm)"
- elif parameters == "time":
+ message = ""
+ arguments = parameters.split()
+ if not parameters: message = "What do you want to show?"
+ elif arguments[0] == "time":
message = universe.categories["internal"]["counters"].get("elapsed") + " increments elapsed since the world was created."
- elif parameters: message = "I don't know what \"" + parameters + "\" is."
- else: message = "What do you want to show?"
- user.send(message)
+ elif arguments[0] == "categories":
+ message = "These are the element categories:$(eol)"
+ categories = universe.categories.keys()
+ categories.sort()
+ for category in categories: message += "$(eol) $(grn)" + category + "$(nrm)"
+ elif arguments[0] == "files":
+ message = "These are the current files containing the universe:$(eol)"
+ filenames = universe.files.keys()
+ filenames.sort()
+ for filename in filenames:
+ if universe.files[filename].is_writeable(): status = "rw"
+ else: status = "ro"
+ message += "$(eol) $(red)(" + status + ") $(grn)" + filename + "$(nrm)"
+ elif arguments[0] == "category":
+ if len(arguments) != 2: message = "You must specify one category."
+ elif arguments[1] in universe.categories:
+ message = "These are the elements in the \"" + arguments[1] + "\" category:$(eol)"
+ elements = [(universe.categories[arguments[1]][x].key) for x in universe.categories[arguments[1]].keys()]
+ elements.sort()
+ for element in elements:
+ message += "$(eol) $(grn)" + element + "$(nrm)"
+ else: message = "Category \"" + arguments[1] + "\" does not exist."
+ elif arguments[0] == "file":
+ if len(arguments) != 2: message = "You must specify one file."
+ elif arguments[1] in universe.files:
+ message = "These are the elements in the \"" + arguments[1] + "\" file:$(eol)"
+ elements = universe.files[arguments[1]].data.sections()
+ elements.sort()
+ for element in elements:
+ message += "$(eol) $(grn)" + element + "$(nrm)"
+ else: message = "Category \"" + arguments[1] + "\" does not exist."
+ elif arguments[0] == "element":
+ if len(arguments) != 2: message = "You must specify one element."
+ elif arguments[1] in universe.contents:
+ element = universe.contents[arguments[1]]
+ message = "These are the properties of the \"" + arguments[1] + "\" element (in \"" + element.origin.filename + "\"):$(eol)"
+ facets = element.facets()
+ facets.sort()
+ for facet in facets:
+ message += "$(eol) $(grn)" + facet + ": $(red)" + escape_macros(element.get(facet)) + "$(nrm)"
+ else: message = "Element \"" + arguments[1] + "\" does not exist."
+ elif arguments[0] == "result":
+ if len(arguments) < 2: message = "You need to specify an expression."
+ else:
+ try:
+ message = repr(eval(" ".join(arguments[1:])))
+ except:
+ message = "Your expression raised an exception!"
+ elif arguments[0] == "log":
+ if len(arguments) == 4:
+ if match("^\d+$", arguments[3]) and int(arguments[3]) >= 0:
+ stop = int(arguments[3])
+ else: stop = -1
+ else: stop = 0
+ if len(arguments) >= 3:
+ if match("^\d+$", arguments[2]) and int(arguments[2]) > 0:
+ start = int(arguments[2])
+ else: start = -1
+ else: start = 10
+ if len(arguments) >= 2:
+ if match("^\d+$", arguments[1]) and 0 <= int(arguments[1]) <= 9:
+ level = int(arguments[1])
+ else: level = -1
+ else: level = 1
+ if level > -1 and start > -1 and stop > -1:
+ message = get_loglines(level, start, stop)
+ else: message = "When specified, level must be 0-9 (default 1), start and stop must be >=1 (default 10 and 1)."
+ else: message = "I don't know what \"" + parameters + "\" is."
+ actor.send(message)
+
+def command_create(actor, parameters):
+ """Create an element if it does not exist."""
+ if not parameters: message = "You must at least specify an element to create."
+ elif not actor.owner: message = ""
+ else:
+ arguments = parameters.split()
+ if len(arguments) == 1: arguments.append("")
+ if len(arguments) == 2:
+ element, filename = arguments
+ if element in universe.contents: message = "The \"" + element + "\" element already exists."
+ else:
+ message = "You create \"" + element + "\" within the universe."
+ logline = actor.owner.account.get("name") + " created an element: " + element
+ if filename:
+ logline += " in file " + filename
+ if filename not in universe.files:
+ message += " Warning: \"" + filename + "\" is not yet included in any other file and will not be read on startup unless this is remedied."
+ Element(element, universe, filename)
+ log(logline, 6)
+ elif len(arguments) > 2: message = "You can only specify an element and a filename."
+ actor.send(message)
+
+def command_destroy(actor, parameters):
+ """Destroy an element if it exists."""
+ if actor.owner:
+ if not parameters: message = "You must specify an element to destroy."
+ else:
+ if parameters not in universe.contents: message = "The \"" + parameters + "\" element does not exist."
+ else:
+ universe.contents[parameters].destroy()
+ message = "You destroy \"" + parameters + "\" within the universe."
+ log(actor.owner.account.get("name") + " destroyed an element: " + parameters, 6)
+ actor.send(message)
+
+def command_set(actor, parameters):
+ """Set a facet of an element."""
+ if not parameters: message = "You must specify an element, a facet and a value."
+ else:
+ arguments = parameters.split(" ", 2)
+ if len(arguments) == 1: message = "What facet of element \"" + arguments[0] + "\" would you like to set?"
+ elif len(arguments) == 2: message = "What value would you like to set for the \"" + arguments[1] + "\" facet of the \"" + arguments[0] + "\" element?"
+ else:
+ element, facet, value = arguments
+ if element not in universe.contents: message = "The \"" + element + "\" element does not exist."
+ else:
+ universe.contents[element].set(facet, value)
+ message = "You have successfully (re)set the \"" + facet + "\" facet of element \"" + element + "\". Try \"show element " + element + "\" for verification."
+ actor.send(message)
+
+def command_delete(actor, parameters):
+ """Delete a facet from an element."""
+ if not parameters: message = "You must specify an element and a facet."
+ else:
+ arguments = parameters.split(" ")
+ if len(arguments) == 1: message = "What facet of element \"" + arguments[0] + "\" would you like to delete?"
+ elif len(arguments) != 2: message = "You may only specify an element and a facet."
+ else:
+ element, facet = arguments
+ if element not in universe.contents: message = "The \"" + element + "\" element does not exist."
+ elif facet not in universe.contents[element].facets(): message = "The \"" + element + "\" element has no \"" + facet + "\" facet."
+ else:
+ universe.contents[element].remove_facet(facet)
+ message = "You have successfully deleted the \"" + facet + "\" facet of element \"" + element + "\". Try \"show element " + element + "\" for verification."
+ actor.send(message)
-def command_error(user, command="", parameters=""):
+def command_error(actor, input_data):
"""Generic error for an unrecognized command word."""
# 90% of the time use a generic error
if randrange(10):
- message = "I'm not sure what \"" + command
- if parameters:
- message += " " + parameters
- message += "\" means..."
+ message = "I'm not sure what \"" + input_data + "\" means..."
# 10% of the time use the classic diku error
else:
message = "Arglebargle, glop-glyf!?!"
# send the error message
- user.send(message)
+ actor.send(message)
+
+def daemonize():
+ """Fork and disassociate from everything."""
+ if universe.contents["internal:process"].getboolean("daemon"):
+ import sys
+ log("Disassociating from the controlling terminal.")
+ if fork(): _exit(0)
+ setsid()
+ for stdpipe in range(3): close(stdpipe)
+ sys.stdin = sys.__stdin__ = file("/dev/null", "r")
+ sys.stdout = sys.stderr = sys.__stdout__ = sys.__stderr__ = file("/dev/null", "w")
# if there is no universe, create an empty one
if not "universe" in locals(): universe = Universe()
projects / mudpy.git / blobdiff