# to use, copy, modify, and distribute this software is granted under
# terms provided in the LICENSE file distributed with this software.
+import codecs
+import ctypes
+import ctypes.util
+import os
+import random
+import re
+import signal
+import socket
+import sys
+import syslog
+import time
+import traceback
+import unicodedata
+
+import mudpy
+
class Element:
def __init__(self, key, universe, filename=None):
"""Set up a new element."""
- import mudpy.data
- import os.path
# keep track of our key name
self.key = key
def getlist(self, facet, default=None):
"""Return values as list type."""
- import mudpy.data
if default is None:
default = []
value = self.get(facet)
def getdict(self, facet, default=None):
"""Return values as dict type."""
- import mudpy.data
if default is None:
default = {}
value = self.get(facet)
def portals(self):
"""Map the portal directions for a room to neighbors."""
- import re
portals = {}
if re.match("""^location:-?\d+,-?\d+,-?\d+$""", self.key):
coordinates = [(int(x))
def __init__(self, filename="", load=False):
"""Initialize the universe."""
- import os
- import os.path
self.categories = {}
self.contents = {}
self.default_origins = {}
def load(self):
"""Load universe data from persistent storage."""
- import mudpy.data
# the files dict must exist and filename needs to be read-only
if not hasattr(
def initialize_server_socket(self):
"""Create and open the listening socket."""
- import socket
# need to know the local address and port number for the listener
host = self.categories["internal"]["network"].get("host")
def __init__(self):
"""Default values for the in-memory user variables."""
- import mudpy.telnet
self.account = None
self.address = ""
self.authenticated = False
def adjust_echoing(self):
"""Adjust echoing to match state menu requirements."""
- import mudpy.telnet
if mudpy.telnet.is_enabled(self, mudpy.telnet.TELOPT_ECHO,
mudpy.telnet.US):
if menu_echo_on(self.state):
prepend_padding=True
):
"""Send arbitrary text to a connected user."""
- import mudpy.telnet
# unless raw mode is on, clean it up all nice and pretty
if not raw:
def enqueue_input(self):
"""Process and enqueue any new input."""
- import mudpy.telnet
- import unicodedata
# check for some input
try:
def log(message, level=0):
"""Log a message."""
- import codecs
- import os.path
- import syslog
- import time
# a couple references we need
file_name = universe.categories["internal"]["logging"].get("file")
def glyph_columns(character):
"""Convenience function to return the column width of a glyph."""
- import unicodedata
if unicodedata.east_asian_width(character) in "FW":
return 2
else:
def wrap_ansi_text(text, width):
"""Wrap text with arbitrary width while ignoring ANSI colors."""
- import unicodedata
# the current position in the entire text string, including all
# characters, printable or otherwise
def weighted_choice(data):
"""Takes a dict weighted by value and returns a random key."""
- import random
# this will hold our expanded list of keys from the data
expanded = []
def random_name():
"""Returns a random character name."""
- import random
# the vowels and consonants needed to create romaji syllables
vowels = [
def replace_macros(user, text, is_input=False):
"""Replaces macros in text output."""
- import codecs
- import mudpy.data
- import os.path
# third person pronouns
pronouns = {
def on_pulse():
"""The things which should happen on each pulse, aside from reloads."""
- import time
# open the listening socket if it hasn't been already
if not hasattr(universe, "listening_socket"):
def check_for_connection(listening_socket):
"""Check for a waiting connection and return a new user object."""
- import mudpy.telnet
# try to accept a new connection
try:
def handle_user_input(user):
"""The main handler, branches to a state-specific handler."""
- import mudpy.telnet
# if the user's client echo is off, send a blank line for aesthetics
if mudpy.telnet.is_enabled(user, mudpy.telnet.TELOPT_ECHO,
def handler_checking_password(user):
"""Handle the login account password."""
- import mudpy.password
# get the next waiting line of input
input_data = user.input_queue.pop(0)
def handler_entering_new_password(user):
"""Handle a new password entry."""
- import mudpy.password
# get the next waiting line of input
input_data = user.input_queue.pop(0)
def handler_verifying_new_password(user):
"""Handle the re-entered new password for verification."""
- import mudpy.password
# get the next waiting line of input
input_data = user.input_queue.pop(0)
def command_say(actor, parameters):
"""Speak to others in the same room."""
- import unicodedata
# check for replacement macros and escape them
parameters = escape_macros(parameters)
def command_show(actor, parameters):
"""Show program data."""
- import re
message = ""
arguments = parameters.split()
if not parameters:
def command_error(actor, input_data):
"""Generic error for an unrecognized command word."""
- import random
# 90% of the time use a generic error
if random.randrange(10):
def daemonize(universe):
"""Fork and disassociate from everything."""
- import codecs
- import ctypes
- import ctypes.util
- import os
- import os.path
- import sys
# only if this is what we're configured to do
if universe.contents["internal:process"].getboolean("daemon"):
def create_pidfile(universe):
"""Write a file containing the current process ID."""
- import codecs
- import os
- import os.path
pid = str(os.getpid())
log("Process ID: " + pid)
file_name = universe.contents["internal:process"].get("pidfile")
def remove_pidfile(universe):
"""Remove the file containing the current process ID."""
- import os
- import os.path
file_name = universe.contents["internal:process"].get("pidfile")
if file_name:
if not os.path.isabs(file_name):
def excepthook(excepttype, value, tracebackdata):
"""Handle uncaught exceptions."""
- import traceback
# assemble the list of errors into a single string
message = "".join(
def sighook(what, where):
"""Handle external signals."""
- import signal
# a generic message
message = "Caught signal: "
def override_excepthook():
"""Redefine sys.excepthook with our own."""
- import sys
sys.excepthook = excepthook
def assign_sighook():
"""Assign a customized handler for some signals."""
- import signal
signal.signal(signal.SIGHUP, sighook)
signal.signal(signal.SIGTERM, sighook)
def setup():
"""This contains functions to be performed when starting the engine."""
- import sys
# see if a configuration file was specified
if len(sys.argv) > 1:
# to use, copy, modify, and distribute this software is granted under
# terms provided in the LICENSE file distributed with this software.
+import base64
+import hashlib
+import math
+import random
+import re
+import struct
+
# convenience constants for indexing the supported hashing algorithms,
# guaranteed a stable part of the interface
MD5 = 0 # hashlib.md5
This is a wrapper around struct.pack, used to turn a list of integers
between 0 and 255 into a packed sequence akin to a C-style string.
"""
- import struct
packed = b""
for number in numbers:
number = int(number)
This is a wrapper around base64.b64encode with preferences
appropriate for encoding Unix-style passwd hash strings.
"""
- import base64
return base64.b64encode(
byte_sequence,
b"./"
need and discard any excess characters over the specified length.
This ensures full distribution over each character of the salt.
"""
- import math
- import random
salt = []
for i in range(int(math.ceil(salt_len * 0.75))):
salt.append(random.randint(0, 255))
facets to this function, a conforming new-style password hash will be
returned.
"""
- import re
assert re.match("^[0-9a-f]{32}$",
legacy_hash), "Not a valid MD5 hexdigest"
collapsed = b""
create(password, algorithm=SHA256, rounds=12, salt_len=16)
"""
- import hashlib
# if a specific salt wasn't specified, we need to generate one
if not salt:
comes out the same as the encoded_hash.
"""
sep = encoded_hash[0]
- import mudpy.misc
algorithm, rounds, salt, hashed = encoded_hash.split(sep)[1:]
if encoded_hash == create(
password=password,
projects / mudpy.git / commitdiff