Andreas/evennia
1
0
Fork 0
mirror of https://github.com/evennia/evennia.git synced 2026-03-28 02:36:32 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'auditing' of https://github.com/strikaco/evennia-dev into strikaco-auditing

Browse source
This commit is contained in:
Griatch 2018-09-29 13:33:28 +02:00
commit 85114d6de5
5 changed files with 453 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

67
evennia/contrib/auditing/README.md Normal file
View file

@ -0,0 +1,67 @@
# Input/Output Auditing
Contrib - Johnny 2017
This is a tap that optionally intercepts all data sent to/from clients and the
server and passes it to a callback of your choosing.
It is intended for quality assurance, post-incident investigations and debugging
but obviously can be abused. All data is recorded in cleartext. Please
be ethical, and if you are unwilling to properly deal with the implications of
recording user passwords or private communications, please do not enable
this module.
Some checks have been implemented to protect the privacy of users.
Files included in this module:
outputs.py - Example callback methods. This module ships with examples of
callbacks that send data as JSON to a file in your game/server/logs
dir or to your native Linux syslog daemon. You can of course write
your own to do other things like post them to Kafka topics.
server.py - Extends the Evennia ServerSession object to pipe data to the
callback upon receipt.
tests.py - Unit tests that check to make sure commands with sensitive
arguments are having their PII scrubbed.
Installation/Configuration:
Deployment is completed by configuring a few settings in server.conf. In short,
you must tell Evennia to use this ServerSession instead of its own, specify
which direction(s) you wish to record and where you want the data sent.
SERVER_SESSION_CLASS = 'evennia.contrib.auditing.server.AuditedServerSession'
# Where to send logs? Define the path to a module containing your callback
# function. It should take a single dict argument as input.
AUDIT_CALLBACK = 'evennia.contrib.auditing.outputs.to_file'
# Log user input? Be ethical about this; it will log all private and
# public communications between players and/or admins.
AUDIT_IN = True/False
# Log server output? This will result in logging of ALL system
# messages and ALL broadcasts to connected players, so on a busy game any
# broadcast to all users will yield a single event for every connected user!
AUDIT_OUT = True/False
# The default output is a dict. Do you want to allow key:value pairs with
# null/blank values? If you're just writing to disk, disabling this saves
# some disk space, but whether you *want* sparse values or not is more of a
# consideration if you're shipping logs to a NoSQL/schemaless database.
AUDIT_ALLOW_SPARSE = True/False
# If you write custom commands that handle sensitive data like passwords,
# you must write a regular expression to remove that before writing to log.
# AUDIT_MASKS is a list of dictionaries that define the names of commands
# and the regexes needed to scrub them.
#
# The sensitive data itself must be captured in a named group with a
# label of 'secret'.
AUDIT_MASKS = [
{'authentication': r"^@auth\s+(?P<secret>[\w]+)"},
]

0
evennia/contrib/auditing/__init__.py Normal file
View file

58
evennia/contrib/auditing/outputs.py Normal file
View file

@ -0,0 +1,58 @@
"""
Auditable Server Sessions - Example Outputs
Example methods demonstrating output destinations for logs generated by
audited server sessions.
This is designed to be a single source of events for developers to customize
and add any additional enhancements before events are written out-- i.e. if you
want to keep a running list of what IPs a user logs in from on account/character
objects, or if you want to perform geoip or ASN lookups on IPs before committing,
or tag certain events with the results of a reputational lookup, this should be
the easiest place to do it. Write a method and invoke it via
`settings.AUDIT_CALLBACK` to have log data objects passed to it.
Evennia contribution - Johnny 2017
"""
from evennia.utils.logger import log_file
import json
import syslog
def to_file(data):
"""
Writes dictionaries of data generated by an AuditedServerSession to files
in JSON format, bucketed by date.
Uses Evennia's native logger and writes to the default
log directory (~/yourgame/server/logs/ or settings.LOG_DIR)
Args:
data (dict): Parsed session transmission data.
"""
# Bucket logs by day and remove objects before serialization
bucket = data.pop('objects')['time'].strftime('%Y-%m-%d')
# Write it
log_file(json.dumps(data), filename="audit_%s.log" % bucket)
def to_syslog(data):
"""
Writes dictionaries of data generated by an AuditedServerSession to syslog.
Takes advantage of your system's native logger and writes to wherever
you have it configured, which is independent of Evennia.
Linux systems tend to write to /var/log/syslog.
If you're running rsyslog, you can configure it to dump and/or forward logs
to disk and/or an external data warehouse (recommended-- if your server is
compromised or taken down, losing your logs along with it is no help!).
Args:
data (dict): Parsed session transmission data.
"""
# Remove objects before serialization
data.pop('objects')
# Write it out
syslog.syslog(json.dumps(data))

233
evennia/contrib/auditing/server.py Normal file
View file

@ -0,0 +1,233 @@
"""
Auditable Server Sessions:
Extension of the stock ServerSession that yields objects representing
user inputs and system outputs.
Evennia contribution - Johnny 2017
"""
import os
import re
import socket
from django.utils import timezone
from django.conf import settings as ev_settings
from evennia.utils import utils, logger, mod_import, get_evennia_version
from evennia.server.serversession import ServerSession
# Attributes governing auditing of commands and where to send log objects
AUDIT_CALLBACK = getattr(ev_settings, 'AUDIT_CALLBACK', None)
AUDIT_IN = getattr(ev_settings, 'AUDIT_IN', False)
AUDIT_OUT = getattr(ev_settings, 'AUDIT_OUT', False)
AUDIT_ALLOW_SPARSE = getattr(ev_settings, 'AUDIT_ALLOW_SPARSE', False)
AUDIT_MASKS = [
{'connect': r"^[@\s]*[connect]{5,8}\s+(\".+?\"|[^\s]+)\s+(?P<secret>.+)"},
{'connect': r"^[@\s]*[connect]{5,8}\s+(?P<secret>[\w]+)"},
{'create': r"^[^@]?[create]{5,6}\s+(\w+|\".+?\")\s+(?P<secret>[\w]+)"},
{'create': r"^[^@]?[create]{5,6}\s+(?P<secret>[\w]+)"},
{'userpassword': r"^[@\s]*[userpassword]{11,14}\s+(\w+|\".+?\")\s+=*\s*(?P<secret>[\w]+)"},
{'userpassword': r"^.*new password set to '(?P<secret>[^']+)'\."},
{'userpassword': r"^.* has changed your password to '(?P<secret>[^']+)'\."},
{'password': r"^[@\s]*[password]{6,9}\s+(?P<secret>.*)"},
] + getattr(ev_settings, 'AUDIT_MASKS', [])
if AUDIT_CALLBACK:
try:
AUDIT_CALLBACK = getattr(mod_import('.'.join(AUDIT_CALLBACK.split('.')[:-1])), AUDIT_CALLBACK.split('.')[-1])
logger.log_info("Auditing module online.")
logger.log_info("Recording user input: %s" % AUDIT_IN)
logger.log_info("Recording server output: %s" % AUDIT_OUT)
logger.log_info("Recording sparse values: %s" % AUDIT_ALLOW_SPARSE)
logger.log_info("Log callback destination: %s" % AUDIT_CALLBACK)
except Exception as e:
logger.log_err("Failed to activate Auditing module. %s" % e)
class AuditedServerSession(ServerSession):
"""
This particular implementation parses all server inputs and/or outputs and
passes a dict containing the parsed metadata to a callback method of your
creation. This is useful for recording player activity where necessary for
security auditing, usage analysis or post-incident forensic discovery.
*** WARNING ***
All strings are recorded and stored in plaintext. This includes those strings
which might contain sensitive data (create, connect, @password). These commands
have their arguments masked by default, but you must mask or mask any
custom commands of your own that handle sensitive information.
See README.md for installation/configuration instructions.
"""
def audit(self, **kwargs):
"""
Extracts messages and system data from a Session object upon message
send or receive.
Kwargs:
src (str): Source of data; 'client' or 'server'. Indicates direction.
text (str or list): Client sends messages to server in the form of
lists. Server sends messages to client as string.
Returns:
log (dict): Dictionary object containing parsed system and user data
related to this message.
"""
# Get time at start of processing
time_obj = timezone.now()
time_str = str(time_obj)
session = self
src = kwargs.pop('src', '?')
bytecount = 0
# Do not log empty lines
if not kwargs: return {}
# Get current session's IP address
client_ip = session.address
# Capture Account name and dbref together
account = session.get_account()
account_token = ''
if account:
account_token = '%s%s' % (account.key, account.dbref)
# Capture Character name and dbref together
char = session.get_puppet()
char_token = ''
if char:
char_token = '%s%s' % (char.key, char.dbref)
# Capture Room name and dbref together
room = None
room_token = ''
if char:
room = char.location
room_token = '%s%s' % (room.key, room.dbref)
# Try to compile an input/output string
def drill(obj, bucket):
if isinstance(obj, dict): return bucket
elif utils.is_iter(obj):
for sub_obj in obj:
bucket.extend(drill(sub_obj, []))
else:
bucket.append(obj)
return bucket
text = kwargs.pop('text', '')
if utils.is_iter(text):
text = '|'.join(drill(text, []))
# Mask any PII in message, where possible
bytecount = len(text.encode('utf-8'))
text = self.mask(text)
# Compile the IP, Account, Character, Room, and the message.
log = {
'time': time_str,
'hostname': socket.getfqdn(),
'application': '%s' % ev_settings.SERVERNAME,
'version': get_evennia_version(),
'pid': os.getpid(),
'direction': 'SND' if src == 'server' else 'RCV',
'protocol': self.protocol_key,
'ip': client_ip,
'session': 'session#%s' % self.sessid,
'account': account_token,
'character': char_token,
'room': room_token,
'text': text.strip(),
'bytes': bytecount,
'data': kwargs,
'objects': {
'time': time_obj,
'session': self,
'account': account,
'character': char,
'room': room,
}
}
# Remove any keys with blank values
if AUDIT_ALLOW_SPARSE == False:
log['data'] = {k:v for k,v in log['data'].iteritems() if v}
log['objects'] = {k:v for k,v in log['objects'].iteritems() if v}
log = {k:v for k,v in log.iteritems() if v}
return log
def mask(self, msg):
"""
Masks potentially sensitive user information within messages before
writing to log. Recording cleartext password attempts is bad policy.
Args:
msg (str): Raw text string sent from client <-> server
Returns:
msg (str): Text string with sensitive information masked out.
"""
# Check to see if the command is embedded within server output
_msg = msg
is_embedded = False
match = re.match(".*Command.*'(.+)'.*is not available.*", msg, flags=re.IGNORECASE)
if match:
msg = match.group(1).replace('\\', '')
submsg = msg
is_embedded = True
for mask in AUDIT_MASKS:
for command, regex in mask.iteritems():
try:
match = re.match(regex, msg, flags=re.IGNORECASE)
except Exception as e:
logger.log_err(regex)
logger.log_err(e)
continue
if match:
term = match.group('secret')
masked = re.sub(term, '*' * len(term.zfill(8)), msg)
if is_embedded:
msg = re.sub(submsg, '%s <Masked: %s>' % (masked, command), _msg, flags=re.IGNORECASE)
else: msg = masked
return msg
return _msg
def data_out(self, **kwargs):
"""
Generic hook for sending data out through the protocol.
Kwargs:
kwargs (any): Other data to the protocol.
"""
if AUDIT_CALLBACK and AUDIT_OUT:
try:
log = self.audit(src='server', **kwargs)
if log: AUDIT_CALLBACK(log)
except Exception as e:
logger.log_err(e)
super(AuditedServerSession, self).data_out(**kwargs)
def data_in(self, **kwargs):
"""
Hook for protocols to send incoming data to the engine.
Kwargs:
kwargs (any): Other data from the protocol.
"""
if AUDIT_CALLBACK and AUDIT_IN:
try:
log = self.audit(src='client', **kwargs)
if log: AUDIT_CALLBACK(log)
except Exception as e:
logger.log_err(e)
super(AuditedServerSession, self).data_in(**kwargs)

95
evennia/contrib/auditing/tests.py Normal file
View file

@ -0,0 +1,95 @@
"""
Module containing the test cases for the Audit system.
"""
from django.conf import settings
from evennia.contrib.auditing.server import AuditedServerSession
from evennia.utils.test_resources import EvenniaTest
import re
# Configure session auditing settings
settings.AUDIT_CALLBACK = "evennia.contrib.auditing.outputs.to_syslog"
settings.AUDIT_IN = True
settings.AUDIT_OUT = True
settings.AUDIT_ALLOW_SPARSE = True
# Configure settings to use custom session
settings.SERVER_SESSION_CLASS = "evennia.contrib.auditing.server.AuditedServerSession"
class AuditingTest(EvenniaTest):
def test_mask(self):
"""
Make sure the 'mask' function is properly masking potentially sensitive
information from strings.
"""
safe_cmds = (
'/say hello to my little friend',
'@ccreate channel = for channeling',
'@create/drop some stuff',
'@create rock',
'@create a pretty shirt : evennia.contrib.clothing.Clothing',
'@charcreate johnnyefhiwuhefwhef',
'Command "@logout" is not available. Maybe you meant "@color" or "@cboot"?',
'/me says, "what is the password?"',
'say the password is plugh',
# Unfortunately given the syntax, there is no way to discern the
# latter of these as sensitive
'@create pretty sunset'
'@create johnny password123',
'{"text": "Command \'do stuff\' is not available. Type \"help\" for help."}'
)
for cmd in safe_cmds:
self.assertEqual(self.session.mask(cmd), cmd)
unsafe_cmds = (
("something - new password set to 'asdfghjk'.", "something - new password set to '********'."),
("someone has changed your password to 'something'.", "someone has changed your password to '*********'."),
('connect johnny password123', 'connect johnny ***********'),
('concnct johnny password123', 'concnct johnny ***********'),
('concnct johnnypassword123', 'concnct *****************'),
('connect "johnny five" "password 123"', 'connect "johnny five" **************'),
('connect johnny "password 123"', 'connect johnny **************'),
('create johnny password123', 'create johnny ***********'),
('@password password1234 = password2345', '@password ***************************'),
('@password password1234 password2345', '@password *************************'),
('@passwd password1234 = password2345', '@passwd ***************************'),
('@userpassword johnny = password234', '@userpassword johnny = ***********'),
('craete johnnypassword123', 'craete *****************'),
("Command 'conncect teddy teddy' is not available. Maybe you meant \"@encode\"?", 'Command \'conncect ******** ********\' is not available. Maybe you meant "@encode"?'),
("{'text': u'Command \\'conncect jsis dfiidf\\' is not available. Type \"help\" for help.'}", "{'text': u'Command \\'conncect jsis ********\\' is not available. Type \"help\" for help.'}")
)
for index, (unsafe, safe) in enumerate(unsafe_cmds):
self.assertEqual(re.sub(' <Masked: .+>', '', self.session.mask(unsafe)).strip(), safe)
# Make sure scrubbing is not being abused to evade monitoring
secrets = [
'say password password password; ive got a secret that i cant explain',
'whisper johnny = password\n let\'s lynch the landlord',
'say connect johnny password1234|the secret life of arabia',
"@password eval(\"__import__('os').system('clear')\", {'__builtins__':{}})"
]
for secret in secrets:
self.assertEqual(self.session.mask(secret), secret)
def test_audit(self):
"""
Make sure the 'audit' function is returning a dictionary based on values
parsed from the Session object.
"""
log = self.session.audit(src='client', text=[['hello']])
obj = {k:v for k,v in log.iteritems() if k in ('direction', 'protocol', 'application', 'text')}
self.assertEqual(obj, {
'direction': 'RCV',
'protocol': 'telnet',
'application': 'Evennia',
'text': 'hello'
})
# Make sure OOB data is being recorded
log = self.session.audit(src='client', text="connect johnny password123", prompt="hp=20|st=10|ma=15", pane=2)
self.assertEqual(log['text'], 'connect johnny ***********')
self.assertEqual(log['data']['prompt'], 'hp=20|st=10|ma=15')
self.assertEqual(log['data']['pane'], 2)