396861feb1
At Ksplice we used /usr/bin/python because we shipped dependencies as Debian / Red Hat packages, which would be installed against the system Python. We were also very careful to use only Python 2.3 features so that even old system Python would still work. None of that is true at Humbug. We expect users to install dependencies themselves, so it's more likely that the Python in $PATH is correct. On OS X in particular, it's common to have five broken Python installs and there's no expectation that /usr/bin/python is the right one. The files which aren't marked executable are not interesting to run as scripts, so we just remove the line there. (In general it's common to have libraries that can also be executed, to run test cases or whatever, but that's not the case here.) (imported from commit 437d4aee2c6e66601ad3334eefd50749cce2eca6)
967 lines
38 KiB
Python
Executable file
967 lines
38 KiB
Python
Executable file
#!/usr/bin/env python
|
|
# Copyright (C) 2012 Humbug, Inc.
|
|
#
|
|
# Permission is hereby granted, free of charge, to any person
|
|
# obtaining a copy of this software and associated documentation files
|
|
# (the "Software"), to deal in the Software without restriction,
|
|
# including without limitation the rights to use, copy, modify, merge,
|
|
# publish, distribute, sublicense, and/or sell copies of the Software,
|
|
# and to permit persons to whom the Software is furnished to do so,
|
|
# subject to the following conditions:
|
|
#
|
|
# The above copyright notice and this permission notice shall be
|
|
# included in all copies or substantial portions of the Software.
|
|
#
|
|
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
|
|
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
|
|
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
|
|
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
|
|
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
|
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
|
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
|
# SOFTWARE.
|
|
|
|
import sys
|
|
import simplejson
|
|
import re
|
|
import time
|
|
import subprocess
|
|
import optparse
|
|
import os
|
|
import datetime
|
|
import textwrap
|
|
import signal
|
|
import logging
|
|
import hashlib
|
|
import tempfile
|
|
import random
|
|
|
|
class CountingBackoff(object):
|
|
def __init__(self, maximum_retries=10):
|
|
self.number_of_retries = 0
|
|
self.maximum_retries = maximum_retries
|
|
|
|
def keep_going(self):
|
|
return self.number_of_retries < self.maximum_retries
|
|
|
|
def succeed(self):
|
|
self.number_of_retries = 0
|
|
|
|
def fail(self):
|
|
self.number_of_retries = min(self.number_of_retries + 1,
|
|
self.maximum_retries)
|
|
|
|
class RandomExponentialBackoff(CountingBackoff):
|
|
def fail(self):
|
|
self.number_of_retries = min(self.number_of_retries + 1,
|
|
self.maximum_retries)
|
|
# Exponential growth with ratio sqrt(2); compute random delay
|
|
# between x and 2x where x is growing exponentially
|
|
delay_scale = int(2 ** (self.number_of_retries / 2.0 - 1)) + 1
|
|
delay = delay_scale + random.randint(1, delay_scale)
|
|
message = "Sleeping for %ss [max %s] before retrying." % (delay, delay_scale * 2)
|
|
try:
|
|
logger.warning(message)
|
|
except NameError:
|
|
print message
|
|
time.sleep(delay)
|
|
|
|
DEFAULT_SITE = "https://humbughq.com"
|
|
|
|
class States:
|
|
Startup, HumbugToZephyr, ZephyrToHumbug, ChildSending = range(4)
|
|
CURRENT_STATE = States.Startup
|
|
|
|
def to_humbug_username(zephyr_username):
|
|
if "@" in zephyr_username:
|
|
(user, realm) = zephyr_username.split("@")
|
|
else:
|
|
(user, realm) = (zephyr_username, "ATHENA.MIT.EDU")
|
|
if realm.upper() == "ATHENA.MIT.EDU":
|
|
return user.lower() + "@mit.edu"
|
|
return user.lower() + "|" + realm.upper() + "@mit.edu"
|
|
|
|
def to_zephyr_username(humbug_username):
|
|
(user, realm) = humbug_username.split("@")
|
|
if "|" not in user:
|
|
return user.lower() + "@ATHENA.MIT.EDU"
|
|
match_user = re.match(r'([a-zA-Z0-9_]+)\|(.+)', user)
|
|
if not match_user:
|
|
raise Exception("Could not parse Zephyr realm for cross-realm user %s" % (humbug_username,))
|
|
return match_user.group(1).lower() + "@" + match_user.group(2).upper()
|
|
|
|
# Checks whether the pair of adjacent lines would have been
|
|
# linewrapped together, had they been intended to be parts of the same
|
|
# paragraph. Our check is whether if you move the first word on the
|
|
# 2nd line onto the first line, the resulting line is either (1)
|
|
# significantly shorter than the following line (which, if they were
|
|
# in the same paragraph, should have been wrapped in a way consistent
|
|
# with how the previous line was wrapped) or (2) shorter than 60
|
|
# characters (our assumed minimum linewrapping threshhold for Zephyr)
|
|
# or (3) the first word of the next line is longer than this entire
|
|
# line.
|
|
def different_paragraph(line, next_line):
|
|
words = next_line.split()
|
|
return (len(line + " " + words[0]) < len(next_line) * 0.8 or
|
|
len(line + " " + words[0]) < 50 or
|
|
len(line) < len(words[0]))
|
|
|
|
# Linewrapping algorithm based on:
|
|
# http://gcbenison.wordpress.com/2011/07/03/a-program-to-intelligently-remove-carriage-returns-so-you-can-paste-text-without-having-it-look-awful/
|
|
def unwrap_lines(body):
|
|
lines = body.split("\n")
|
|
result = ""
|
|
previous_line = lines[0]
|
|
for line in lines[1:]:
|
|
line = line.rstrip()
|
|
if (re.match(r'^\W', line, flags=re.UNICODE)
|
|
and re.match(r'^\W', previous_line, flags=re.UNICODE)):
|
|
result += previous_line + "\n"
|
|
elif (line == "" or
|
|
previous_line == "" or
|
|
re.match(r'^\W', line, flags=re.UNICODE) or
|
|
different_paragraph(previous_line, line)):
|
|
# Use 2 newlines to separate sections so that we
|
|
# trigger proper Markdown processing on things like
|
|
# bulleted lists
|
|
result += previous_line + "\n\n"
|
|
else:
|
|
result += previous_line + " "
|
|
previous_line = line
|
|
result += previous_line
|
|
return result
|
|
|
|
def send_humbug(zeph):
|
|
message = {}
|
|
if options.forward_class_messages:
|
|
message["forged"] = "yes"
|
|
message['type'] = zeph['type']
|
|
message['time'] = zeph['time']
|
|
message['sender'] = to_humbug_username(zeph['sender'])
|
|
if "subject" in zeph:
|
|
# Truncate the subject to the current limit in Humbug. No
|
|
# need to do this for stream names, since we're only
|
|
# subscribed to valid stream names.
|
|
message["subject"] = zeph["subject"][:60]
|
|
if zeph['type'] == 'stream':
|
|
# Forward messages sent to -c foo -i bar to stream bar subject "instance"
|
|
if zeph["stream"] == "message":
|
|
message['to'] = zeph['subject'].lower()
|
|
message['subject'] = "instance %s" % (zeph['subject'],)
|
|
elif zeph["stream"] == "tabbott-test5":
|
|
message['to'] = zeph['subject'].lower()
|
|
message['subject'] = "test instance %s" % (zeph['subject'],)
|
|
else:
|
|
message["to"] = zeph["stream"]
|
|
else:
|
|
message["to"] = zeph["recipient"]
|
|
message['content'] = unwrap_lines(zeph['content'])
|
|
|
|
if options.test_mode and options.site == DEFAULT_SITE:
|
|
logger.debug("Message is: %s" % (str(message),))
|
|
return {'result': "success"}
|
|
|
|
return humbug_client.send_message(message)
|
|
|
|
def send_error_humbug(error_msg):
|
|
message = {"type": "private",
|
|
"sender": humbug_account_email,
|
|
"to": humbug_account_email,
|
|
"content": error_msg,
|
|
}
|
|
humbug_client.send_message(message)
|
|
|
|
current_zephyr_subs = set()
|
|
def zephyr_bulk_subscribe(subs):
|
|
try:
|
|
zephyr._z.subAll(subs)
|
|
except IOError:
|
|
# Since we haven't added the subscription to
|
|
# current_zephyr_subs yet, we can just return (so that we'll
|
|
# continue processing normal messages) and we'll end up
|
|
# retrying the next time the bot checks its subscriptions are
|
|
# up to date.
|
|
logger.exception("Error subscribing to streams (will retry automatically):")
|
|
logger.warning("Streams were: %s" % ([cls for cls, instance, recipient in subs],))
|
|
return
|
|
try:
|
|
actual_zephyr_subs = [cls for (cls, _, _) in zephyr._z.getSubscriptions()]
|
|
except IOError:
|
|
logger.exception("Error getting current Zephyr subscriptions")
|
|
# Don't add anything to current_zephyr_subs so that we'll
|
|
# retry the next time we check for streams to subscribe to
|
|
# (within 15 seconds).
|
|
return
|
|
for (cls, instance, recipient) in subs:
|
|
if cls not in actual_zephyr_subs:
|
|
logger.error("Zephyr failed to subscribe us to %s; will retry" % (cls,))
|
|
try:
|
|
# We'll retry automatically when we next check for
|
|
# streams to subscribe to (within 15 seconds), but
|
|
# it's worth doing 1 retry immediately to avoid
|
|
# missing 15 seconds of messages on the affected
|
|
# classes
|
|
zephyr._z.sub(cls, instance, recipient)
|
|
except IOError:
|
|
pass
|
|
else:
|
|
current_zephyr_subs.add(cls)
|
|
|
|
def update_subscriptions():
|
|
try:
|
|
f = file("/home/humbug/public_streams", "r")
|
|
public_streams = simplejson.loads(f.read())
|
|
f.close()
|
|
except:
|
|
logger.exception("Error reading public streams:")
|
|
return
|
|
|
|
classes_to_subscribe = set()
|
|
for stream in public_streams:
|
|
zephyr_class = stream.encode("utf-8")
|
|
if (options.shard is not None and
|
|
not hashlib.sha1(zephyr_class).hexdigest().startswith(options.shard)):
|
|
# This stream is being handled by a different zephyr_mirror job.
|
|
continue
|
|
if zephyr_class in current_zephyr_subs:
|
|
continue
|
|
classes_to_subscribe.add((zephyr_class, "*", "*"))
|
|
|
|
if len(classes_to_subscribe) > 0:
|
|
zephyr_bulk_subscribe(list(classes_to_subscribe))
|
|
|
|
def maybe_kill_child():
|
|
try:
|
|
if child_pid is not None:
|
|
os.kill(child_pid, signal.SIGTERM)
|
|
except OSError:
|
|
# We don't care if the child process no longer exists, so just log the error
|
|
logger.exception("")
|
|
|
|
def maybe_restart_mirroring_script():
|
|
if os.stat(os.path.join(options.root_path, "stamps", "restart_stamp")).st_mtime > start_time or \
|
|
((options.user == "tabbott" or options.user == "tabbott/extra") and
|
|
os.stat(os.path.join(options.root_path, "stamps", "tabbott_stamp")).st_mtime > start_time):
|
|
logger.warning("")
|
|
logger.warning("zephyr mirroring script has been updated; restarting...")
|
|
maybe_kill_child()
|
|
try:
|
|
zephyr._z.cancelSubs()
|
|
except IOError:
|
|
# We don't care whether we failed to cancel subs properly, but we should log it
|
|
logger.exception("")
|
|
while True:
|
|
try:
|
|
os.execvp(os.path.join(options.root_path, "user_root", "zephyr_mirror_backend.py"), sys.argv)
|
|
except Exception:
|
|
logger.exception("Error restarting mirroring script; trying again... Traceback:")
|
|
time.sleep(1)
|
|
|
|
def process_loop(log):
|
|
sleep_count = 0
|
|
sleep_time = 0.1
|
|
while True:
|
|
try:
|
|
notice = zephyr.receive(block=False)
|
|
except Exception:
|
|
logger.exception("Error checking for new zephyrs:")
|
|
time.sleep(1)
|
|
continue
|
|
if notice is not None:
|
|
try:
|
|
process_notice(notice, log)
|
|
except Exception:
|
|
logger.exception("Error relaying zephyr:")
|
|
time.sleep(2)
|
|
|
|
try:
|
|
maybe_restart_mirroring_script()
|
|
except Exception:
|
|
logger.exception("Error checking whether restart is required:")
|
|
|
|
time.sleep(sleep_time)
|
|
sleep_count += sleep_time
|
|
if sleep_count > 15:
|
|
sleep_count = 0
|
|
if options.forward_class_messages:
|
|
# Ask the Humbug server about any new classes to subscribe to
|
|
try:
|
|
update_subscriptions()
|
|
except Exception:
|
|
logger.exception("Error updating subscriptions from Humbug:")
|
|
|
|
def parse_zephyr_body(zephyr_data):
|
|
try:
|
|
(zsig, body) = zephyr_data.split("\x00", 1)
|
|
except ValueError:
|
|
(zsig, body) = ("", zephyr_data)
|
|
return (zsig, body)
|
|
|
|
def process_notice(notice, log):
|
|
(zsig, body) = parse_zephyr_body(notice.message)
|
|
is_personal = False
|
|
is_huddle = False
|
|
|
|
if notice.opcode == "PING":
|
|
# skip PING messages
|
|
return
|
|
|
|
zephyr_class = notice.cls.lower()
|
|
|
|
if notice.recipient != "":
|
|
is_personal = True
|
|
# Drop messages not to the listed subscriptions
|
|
if is_personal and not options.forward_personals:
|
|
return
|
|
if (zephyr_class not in current_zephyr_subs) and not is_personal:
|
|
logger.debug("Skipping ... %s/%s/%s" %
|
|
(zephyr_class, notice.instance, is_personal))
|
|
return
|
|
if notice.format.startswith("Zephyr error: See") or notice.format.endswith("@(@color(blue))"):
|
|
logger.debug("Skipping message we got from Humbug!")
|
|
return
|
|
|
|
if is_personal:
|
|
if body.startswith("CC:"):
|
|
is_huddle = True
|
|
# Map "CC: sipbtest espuser" => "starnine@mit.edu,espuser@mit.edu"
|
|
huddle_recipients = [to_humbug_username(x.strip()) for x in
|
|
body.split("\n")[0][4:].split()]
|
|
if notice.sender not in huddle_recipients:
|
|
huddle_recipients.append(to_humbug_username(notice.sender))
|
|
body = body.split("\n", 1)[1]
|
|
|
|
zeph = { 'time' : str(notice.time),
|
|
'sender' : notice.sender,
|
|
'zsig' : zsig, # logged here but not used by app
|
|
'content' : body }
|
|
if is_huddle:
|
|
zeph['type'] = 'private'
|
|
zeph['recipient'] = huddle_recipients
|
|
elif is_personal:
|
|
zeph['type'] = 'private'
|
|
zeph['recipient'] = to_humbug_username(notice.recipient)
|
|
else:
|
|
zeph['type'] = 'stream'
|
|
zeph['stream'] = zephyr_class
|
|
if notice.instance.strip() != "":
|
|
zeph['subject'] = notice.instance
|
|
else:
|
|
zeph["subject"] = '(instance "%s")' % (notice.instance,)
|
|
|
|
# Add instances in for instanced personals
|
|
if is_personal:
|
|
if notice.cls.lower() != "message" and notice.instance.lower != "personal":
|
|
heading = "[-c %s -i %s]\n" % (notice.cls, notice.instance)
|
|
elif notice.cls.lower() != "message":
|
|
heading = "[-c %s]\n" % (notice.cls,)
|
|
elif notice.instance.lower() != "personal":
|
|
heading = "[-i %s]\n" % (notice.instance,)
|
|
else:
|
|
heading = ""
|
|
zeph["content"] = heading + zeph["content"]
|
|
|
|
zeph = decode_unicode_byte_strings(zeph)
|
|
|
|
logger.info("Received a message on %s/%s from %s..." %
|
|
(zephyr_class, notice.instance, notice.sender))
|
|
if log is not None:
|
|
log.write(simplejson.dumps(zeph) + '\n')
|
|
log.flush()
|
|
|
|
if os.fork() == 0:
|
|
global CURRENT_STATE
|
|
CURRENT_STATE = States.ChildSending
|
|
# Actually send the message in a child process, to avoid blocking.
|
|
try:
|
|
res = send_humbug(zeph)
|
|
if res.get("result") != "success":
|
|
logger.error("Error relaying zephyr:\n%s\n%s" % (zeph, res))
|
|
except Exception:
|
|
logger.exception("Error relaying zephyr:")
|
|
finally:
|
|
os._exit(0)
|
|
|
|
def decode_unicode_byte_strings(zeph):
|
|
for field in zeph.keys():
|
|
if isinstance(zeph[field], str):
|
|
try:
|
|
decoded = zeph[field].decode("utf-8")
|
|
except Exception:
|
|
decoded = zeph[field].decode("iso-8859-1")
|
|
zeph[field] = decoded
|
|
return zeph
|
|
|
|
def quit_failed_initialization(message):
|
|
logger.error(message)
|
|
maybe_kill_child()
|
|
sys.exit(1)
|
|
|
|
def zephyr_init_autoretry():
|
|
backoff = RandomExponentialBackoff()
|
|
while backoff.keep_going():
|
|
try:
|
|
# zephyr.init() tries to clear old subscriptions, and thus
|
|
# sometimes gets a SERVNAK from the server
|
|
zephyr.init()
|
|
backoff.succeed()
|
|
return
|
|
except IOError:
|
|
logger.exception("Error initializing Zephyr library (retrying). Traceback:")
|
|
backoff.fail()
|
|
|
|
quit_failed_initialization("Could not initialize Zephyr library, quitting!")
|
|
|
|
def zephyr_subscribe_autoretry(sub):
|
|
backoff = RandomExponentialBackoff()
|
|
while backoff.keep_going():
|
|
try:
|
|
zephyr.Subscriptions().add(sub)
|
|
backoff.succeed()
|
|
return
|
|
except IOError:
|
|
# Probably a SERVNAK from the zephyr server, but log the
|
|
# traceback just in case it's something else
|
|
logger.exception("Error subscribing to personals (retrying). Traceback:")
|
|
backoff.fail()
|
|
|
|
quit_failed_initialization("Could not subscribe to personals, quitting!")
|
|
|
|
def zephyr_to_humbug(options):
|
|
zephyr_init_autoretry()
|
|
if options.forward_class_messages:
|
|
update_subscriptions()
|
|
if options.forward_personals:
|
|
# Subscribe to personals; we really can't operate without
|
|
# those subscriptions, so just retry until it works.
|
|
zephyr_subscribe_autoretry(("message", "*", "%me%"))
|
|
if subscribed_to_mail_messages():
|
|
zephyr_subscribe_autoretry(("mail", "inbox", "%me%"))
|
|
|
|
if options.resend_log_path is not None:
|
|
with open(options.resend_log_path, 'r') as log:
|
|
for ln in log:
|
|
try:
|
|
zeph = simplejson.loads(ln)
|
|
# New messages added to the log shouldn't have any
|
|
# elements of type str (they should already all be
|
|
# unicode), but older messages in the log are
|
|
# still of type str, so convert them before we
|
|
# send the message
|
|
zeph = decode_unicode_byte_strings(zeph)
|
|
# Handle importing older zephyrs in the logs
|
|
# where it isn't called a "stream" yet
|
|
if "class" in zeph:
|
|
zeph["stream"] = zeph["class"]
|
|
if "instance" in zeph:
|
|
zeph["subject"] = zeph["instance"]
|
|
logger.info("sending saved message to %s from %s..." %
|
|
(zeph.get('stream', zeph.get('recipient')),
|
|
zeph['sender']))
|
|
send_humbug(zeph)
|
|
except Exception:
|
|
logger.exception("Could not send saved zephyr:")
|
|
time.sleep(2)
|
|
|
|
logger.info("Successfully initialized; Starting receive loop.")
|
|
|
|
if options.log_path is not None:
|
|
with open(options.log_path, 'a') as log:
|
|
process_loop(log)
|
|
else:
|
|
process_loop(None)
|
|
|
|
def send_zephyr(zwrite_args, content):
|
|
p = subprocess.Popen(zwrite_args, stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
stdout, stderr = p.communicate(input=content.encode("utf-8"))
|
|
if p.returncode:
|
|
logger.error("zwrite command '%s' failed with return code %d:" % (
|
|
" ".join(zwrite_args), p.returncode,))
|
|
if stdout:
|
|
logger.info("stdout: " + stdout)
|
|
elif stderr:
|
|
logger.warning("zwrite command '%s' printed the following warning:" % (
|
|
" ".join(zwrite_args),))
|
|
if stderr:
|
|
logger.warning("stderr: " + stderr)
|
|
return (p.returncode, stderr)
|
|
|
|
def send_authed_zephyr(zwrite_args, content):
|
|
return send_zephyr(zwrite_args, content)
|
|
|
|
def send_unauthed_zephyr(zwrite_args, content):
|
|
return send_zephyr(zwrite_args + ["-d"], content)
|
|
|
|
def forward_to_zephyr(message):
|
|
wrapper = textwrap.TextWrapper(break_long_words=False, break_on_hyphens=False)
|
|
wrapped_content = "\n".join("\n".join(wrapper.wrap(line))
|
|
for line in message["content"].split("\n"))
|
|
|
|
zwrite_args = ["zwrite", "-n", "-s", zsig_fullname, "-F", "Zephyr error: See http://zephyr.1ts.org/wiki/df"]
|
|
if message['type'] == "stream":
|
|
zephyr_class = message["display_recipient"]
|
|
instance = message["subject"]
|
|
|
|
match_whitespace_instance = re.match(r'^\(instance "(\s*)"\)$', instance)
|
|
if match_whitespace_instance:
|
|
# Forward messages sent to '(instance "WHITESPACE")' back to the
|
|
# appropriate WHITESPACE instance for bidirectional mirroring
|
|
instance = match_whitespace_instance.group(1)
|
|
elif (instance == "instance %s" % (zephyr_class,) or
|
|
instance == "test instance %s" % (zephyr_class,)):
|
|
# Forward messages to e.g. -c -i white-magic back from the
|
|
# place we forward them to
|
|
if instance.startswith("test"):
|
|
instance = zephyr_class
|
|
zephyr_class = "tabbott-test5"
|
|
else:
|
|
instance = zephyr_class
|
|
zephyr_class = "message"
|
|
zwrite_args.extend(["-c", zephyr_class, "-i", instance])
|
|
logger.info("Forwarding message to class %s, instance %s" % (zephyr_class, instance))
|
|
elif message['type'] == "private":
|
|
if len(message['display_recipient']) == 1:
|
|
recipient = to_zephyr_username(message["display_recipient"][0]["email"])
|
|
recipients = [recipient]
|
|
elif len(message['display_recipient']) == 2:
|
|
recipient = ""
|
|
for r in message["display_recipient"]:
|
|
if r["email"].lower() != humbug_account_email.lower():
|
|
recipient = to_zephyr_username(r["email"])
|
|
break
|
|
recipients = [recipient]
|
|
else:
|
|
zwrite_args.extend(["-C"])
|
|
# We drop the @ATHENA.MIT.EDU here because otherwise the
|
|
# "CC: user1 user2 ..." output will be unnecessarily verbose.
|
|
recipients = [to_zephyr_username(user["email"]).replace("@ATHENA.MIT.EDU", "")
|
|
for user in message["display_recipient"]]
|
|
logger.info("Forwarding message to %s" % (recipients,))
|
|
zwrite_args.extend(recipients)
|
|
|
|
if options.test_mode:
|
|
logger.debug("Would have forwarded: %s\n%s" %
|
|
(zwrite_args, wrapped_content.encode("utf-8")))
|
|
return
|
|
|
|
heading = "Hi there! This is an automated message from Humbug."
|
|
support_closing = """If you have any questions, please be in touch through the \
|
|
Feedback tab or at support@humbughq.com."""
|
|
|
|
(code, stderr) = send_authed_zephyr(zwrite_args, wrapped_content)
|
|
if code == 0 and stderr == "":
|
|
return
|
|
elif code == 0:
|
|
return send_error_humbug("""%s
|
|
|
|
Your last message was successfully mirrored to zephyr, but zwrite \
|
|
returned the following warning:
|
|
|
|
%s
|
|
|
|
%s""" % (heading, stderr, support_closing))
|
|
elif code != 0 and (stderr.startswith("zwrite: Ticket expired while sending notice to ") or
|
|
stderr.startswith("zwrite: No credentials cache found while sending notice to ")):
|
|
# Retry sending the message unauthenticated; if that works,
|
|
# just notify the user that they need to renew their tickets
|
|
(code, stderr) = send_unauthed_zephyr(zwrite_args, wrapped_content)
|
|
if code == 0:
|
|
return send_error_humbug("""%s
|
|
|
|
Your last message was forwarded from Humbug to Zephyr unauthenticated, \
|
|
because your Kerberos tickets have expired. It was sent successfully, \
|
|
but please renew your Kerberos tickets in the screen session where you \
|
|
are running the Humbug-Zephyr mirroring bot, so we can send \
|
|
authenticated Zephyr messages for you again.
|
|
|
|
%s""" % (heading, support_closing))
|
|
|
|
# zwrite failed and it wasn't because of expired tickets: This is
|
|
# probably because the recipient isn't subscribed to personals,
|
|
# but regardless, we should just notify the user.
|
|
return send_error_humbug("""%s
|
|
|
|
Your Humbug-Zephyr mirror bot was unable to forward that last message \
|
|
from Humbug to Zephyr. That means that while Humbug users (like you) \
|
|
received it, Zephyr users did not. The error message from zwrite was:
|
|
|
|
%s
|
|
|
|
%s""" % (heading, stderr, support_closing))
|
|
|
|
def maybe_forward_to_zephyr(message):
|
|
if (message["sender_email"] == humbug_account_email):
|
|
if not ((message["type"] == "stream") or
|
|
(message["type"] == "private" and
|
|
False not in [u["email"].lower().endswith("mit.edu") for u in
|
|
message["display_recipient"]])):
|
|
# Don't try forward private messages with non-MIT users
|
|
# to MIT Zephyr.
|
|
return
|
|
timestamp_now = datetime.datetime.now().strftime("%s")
|
|
if float(message["timestamp"]) < float(timestamp_now) - 15:
|
|
logger.warning("Skipping out of order message: %s < %s" %
|
|
(message["timestamp"], timestamp_now))
|
|
return
|
|
try:
|
|
forward_to_zephyr(message)
|
|
except Exception:
|
|
# Don't let an exception forwarding one message crash the
|
|
# whole process
|
|
logger.exception("Error forwarding message:")
|
|
|
|
def humbug_to_zephyr(options):
|
|
# Sync messages from zephyr to humbug
|
|
logger.info("Starting syncing messages.")
|
|
while True:
|
|
try:
|
|
humbug_client.call_on_each_message(maybe_forward_to_zephyr)
|
|
except Exception:
|
|
logger.exception("Error syncing messages:")
|
|
time.sleep(1)
|
|
|
|
def subscribed_to_mail_messages():
|
|
# In case we have lost our AFS tokens and those won't be able to
|
|
# parse the Zephyr subs file, first try reading in result of this
|
|
# query from the environment so we can avoid the filesystem read.
|
|
stored_result = os.environ.get("HUMBUG_FORWARD_MAIL_ZEPHYRS")
|
|
if stored_result is not None:
|
|
return stored_result == "True"
|
|
for (cls, instance, recipient) in parse_zephyr_subs(verbose=False):
|
|
if (cls.lower() == "mail" and instance.lower() == "inbox"):
|
|
os.environ["HUMBUG_FORWARD_MAIL_ZEPHYRS"] = "True"
|
|
return True
|
|
os.environ["HUMBUG_FORWARD_MAIL_ZEPHYRS"] = "False"
|
|
return False
|
|
|
|
def add_humbug_subscriptions(verbose):
|
|
zephyr_subscriptions = set()
|
|
skipped = set()
|
|
for (cls, instance, recipient) in parse_zephyr_subs(verbose=verbose):
|
|
if cls.lower() == "message":
|
|
if recipient != "*":
|
|
# We already have a (message, *, you) subscription, so
|
|
# these are redundant
|
|
continue
|
|
# We don't support subscribing to (message, *)
|
|
if instance == "*":
|
|
if recipient == "*":
|
|
skipped.add((cls, instance, recipient, "subscribing to all of class message is not supported."))
|
|
continue
|
|
# If you're on -i white-magic on zephyr, get on stream white-magic on humbug
|
|
# instead of subscribing to stream "message" on humbug
|
|
zephyr_subscriptions.add(instance)
|
|
continue
|
|
elif cls.lower() == "mail" and instance.lower() == "inbox":
|
|
# We forward mail zephyrs, so no need to log a warning.
|
|
continue
|
|
elif len(cls) > 30:
|
|
skipped.add((cls, instance, recipient, "Class longer than 30 characters"))
|
|
continue
|
|
elif instance != "*":
|
|
skipped.add((cls, instance, recipient, "Unsupported non-* instance"))
|
|
continue
|
|
elif recipient != "*":
|
|
skipped.add((cls, instance, recipient, "Unsupported non-* recipient."))
|
|
continue
|
|
zephyr_subscriptions.add(cls)
|
|
|
|
if len(zephyr_subscriptions) != 0:
|
|
res = humbug_client.add_subscriptions(list(zephyr_subscriptions))
|
|
if res.get("result") != "success":
|
|
logger.error("Error subscribing to streams:\n%s" % (res["msg"],))
|
|
return
|
|
|
|
already = res.get("already_subscribed")
|
|
new = res.get("subscribed")
|
|
if verbose:
|
|
if already is not None and len(already) > 0:
|
|
logger.info("\nAlready subscribed to: %s" % (", ".join(already.values()[0]),))
|
|
if new is not None and len(new) > 0:
|
|
logger.info("\nSuccessfully subscribed to: %s" % (", ".join(new.values()[0]),))
|
|
|
|
if len(skipped) > 0:
|
|
if verbose:
|
|
logger.info("\n" + "\n".join(textwrap.wrap("""\
|
|
You have some lines in ~/.zephyr.subs that could not be
|
|
synced to your Humbug subscriptions because they do not
|
|
use "*" as both the instance and recipient and not one of
|
|
the special cases (e.g. personals and mail zephyrs) that
|
|
Humbug has a mechanism for forwarding. Humbug does not
|
|
allow subscribing to only some subjects on a Humbug
|
|
stream, so this tool has not created a corresponding
|
|
Humbug subscription to these lines in ~/.zephyr.subs:
|
|
""")) + "\n")
|
|
|
|
for (cls, instance, recipient, reason) in skipped:
|
|
if verbose:
|
|
if reason != "":
|
|
logger.info(" [%s,%s,%s] (%s)" % (cls, instance, recipient, reason))
|
|
else:
|
|
logger.info(" [%s,%s,%s]" % (cls, instance, recipient))
|
|
if len(skipped) > 0:
|
|
if verbose:
|
|
logger.info("\n" + "\n".join(textwrap.wrap("""\
|
|
If you wish to be subscribed to any Humbug streams related
|
|
to these .zephyrs.subs lines, please do so via the Humbug
|
|
web interface.
|
|
""")) + "\n")
|
|
if verbose:
|
|
logger.info("\nIMPORTANT: Please reload the Humbug app for these changes to take effect.\n")
|
|
|
|
def valid_stream_name(name):
|
|
return name != ""
|
|
|
|
def parse_zephyr_subs(verbose=False):
|
|
zephyr_subscriptions = set()
|
|
subs_file = os.path.join(os.environ["HOME"], ".zephyr.subs")
|
|
if not os.path.exists(subs_file):
|
|
if verbose:
|
|
logger.error("Couldn't find ~/.zephyr.subs!")
|
|
return []
|
|
|
|
for line in file(subs_file, "r").readlines():
|
|
line = line.strip()
|
|
if len(line) == 0:
|
|
continue
|
|
try:
|
|
(cls, instance, recipient) = line.split(",")
|
|
cls = cls.replace("%me%", options.user)
|
|
instance = instance.replace("%me%", options.user)
|
|
recipient = recipient.replace("%me%", options.user)
|
|
if not valid_stream_name(cls):
|
|
if verbose:
|
|
logger.error("Skipping subscription to unsupported class name: [%s]" % (line,))
|
|
continue
|
|
except Exception:
|
|
if verbose:
|
|
logger.error("Couldn't parse ~/.zephyr.subs line: [%s]" % (line,))
|
|
continue
|
|
zephyr_subscriptions.add((cls.strip(), instance.strip(), recipient.strip()))
|
|
return zephyr_subscriptions
|
|
|
|
def fetch_fullname(username):
|
|
try:
|
|
proc = subprocess.Popen(['hesinfo', username, 'passwd'],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE)
|
|
out, _err_unused = proc.communicate()
|
|
if proc.returncode == 0:
|
|
return out.split(':')[4].split(',')[0]
|
|
except Exception:
|
|
logger.exception("Error getting fullname for %s:" % (username,))
|
|
|
|
return username
|
|
|
|
def open_logger():
|
|
if options.forward_class_messages:
|
|
if options.test_mode:
|
|
log_file = "/home/humbug/test-mirror-log"
|
|
else:
|
|
log_file = "/home/humbug/mirror-log"
|
|
else:
|
|
f = tempfile.NamedTemporaryFile(prefix="humbug-log.%s." % (options.user,),
|
|
delete=False)
|
|
log_file = f.name
|
|
# Close the file descriptor, since the logging system will
|
|
# reopen it anyway.
|
|
f.close()
|
|
logger = logging.getLogger(__name__)
|
|
log_format = "%(asctime)s <initial>: %(message)s"
|
|
formatter = logging.Formatter(log_format)
|
|
logging.basicConfig(format=log_format)
|
|
logger.setLevel(logging.DEBUG)
|
|
file_handler = logging.FileHandler(log_file)
|
|
file_handler.setFormatter(formatter)
|
|
logger.addHandler(file_handler)
|
|
return logger
|
|
|
|
def configure_logger(logger, direction_name):
|
|
if direction_name is None:
|
|
log_format = "%(message)s"
|
|
else:
|
|
log_format = "%(asctime)s [" + direction_name + "] %(message)s"
|
|
formatter = logging.Formatter(log_format)
|
|
|
|
# Replace the formatters for the file and stdout loggers
|
|
for handler in logger.handlers:
|
|
handler.setFormatter(formatter)
|
|
root_logger = logging.getLogger()
|
|
for handler in root_logger.handlers:
|
|
handler.setFormatter(formatter)
|
|
|
|
def parse_args():
|
|
parser = optparse.OptionParser()
|
|
parser.add_option('--forward-class-messages',
|
|
default=False,
|
|
help=optparse.SUPPRESS_HELP,
|
|
action='store_true')
|
|
parser.add_option('--shard',
|
|
help=optparse.SUPPRESS_HELP)
|
|
parser.add_option('--noshard',
|
|
default=False,
|
|
help=optparse.SUPPRESS_HELP,
|
|
action='store_true')
|
|
parser.add_option('--resend-log',
|
|
dest='resend_log_path',
|
|
help=optparse.SUPPRESS_HELP)
|
|
parser.add_option('--enable-log',
|
|
dest='log_path',
|
|
help=optparse.SUPPRESS_HELP)
|
|
parser.add_option('--no-forward-personals',
|
|
dest='forward_personals',
|
|
help=optparse.SUPPRESS_HELP,
|
|
default=True,
|
|
action='store_false')
|
|
parser.add_option('--no-forward-from-humbug',
|
|
default=True,
|
|
dest='forward_from_humbug',
|
|
help=optparse.SUPPRESS_HELP,
|
|
action='store_false')
|
|
parser.add_option('--verbose',
|
|
default=False,
|
|
help=optparse.SUPPRESS_HELP,
|
|
action='store_true')
|
|
parser.add_option('--sync-subscriptions',
|
|
default=False,
|
|
action='store_true')
|
|
parser.add_option('--site',
|
|
default=DEFAULT_SITE,
|
|
help=optparse.SUPPRESS_HELP)
|
|
parser.add_option('--user',
|
|
default=os.environ["USER"],
|
|
help=optparse.SUPPRESS_HELP)
|
|
parser.add_option('--root-path',
|
|
default="/afs/athena.mit.edu/user/t/a/tabbott/for_friends",
|
|
help=optparse.SUPPRESS_HELP)
|
|
parser.add_option('--test-mode',
|
|
default=False,
|
|
help=optparse.SUPPRESS_HELP,
|
|
action='store_true')
|
|
parser.add_option('--api-key-file',
|
|
default=os.path.join(os.environ["HOME"], "Private", ".humbug-api-key"))
|
|
return parser.parse_args()
|
|
|
|
def die_gracefully(signal, frame):
|
|
if CURRENT_STATE == States.HumbugToZephyr or CURRENT_STATE == States.ChildSending:
|
|
# this is a child process, so we want os._exit (no clean-up necessary)
|
|
os._exit(1)
|
|
|
|
if CURRENT_STATE == States.ZephyrToHumbug:
|
|
try:
|
|
# zephyr=>humbug processes may have added subs, so run cancelSubs
|
|
zephyr._z.cancelSubs()
|
|
except IOError:
|
|
# We don't care whether we failed to cancel subs properly, but we should log it
|
|
logger.exception("")
|
|
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
# Set the SIGCHLD handler back to SIG_DFL to prevent these errors
|
|
# when importing the "requests" module after being restarted using
|
|
# the restart_stamp functionality:
|
|
#
|
|
# close failed in file object destructor:
|
|
# IOError: [Errno 10] No child processes
|
|
signal.signal(signal.SIGCHLD, signal.SIG_DFL)
|
|
|
|
signal.signal(signal.SIGINT, die_gracefully)
|
|
|
|
(options, args) = parse_args()
|
|
|
|
logger = open_logger()
|
|
configure_logger(logger, "parent")
|
|
|
|
# The 'api' directory needs to go first, so that 'import humbug' won't pick
|
|
# up some other directory named 'humbug'.
|
|
pyzephyr_lib_path = "python-zephyr/build/lib.linux-" + os.uname()[4] + "-2.6/"
|
|
sys.path[:0] = [os.path.join(options.root_path, 'api'),
|
|
options.root_path,
|
|
os.path.join(options.root_path, "python-zephyr"),
|
|
os.path.join(options.root_path, pyzephyr_lib_path)]
|
|
|
|
# In case this is an automated restart of the mirroring script,
|
|
# and we have lost AFS tokens, first try reading the API key from
|
|
# the environment so that we can skip doing a filesystem read.
|
|
if os.environ.get("HUMBUG_API_KEY") is not None:
|
|
api_key = os.environ.get("HUMBUG_API_KEY")
|
|
else:
|
|
if not os.path.exists(options.api_key_file):
|
|
logger.error("\n" + "\n".join(textwrap.wrap("""\
|
|
Could not find API key file.
|
|
You need to either place your api key file at %s,
|
|
or specify the --api-key-file option.""" % (options.api_key_file,))))
|
|
sys.exit(1)
|
|
api_key = file(options.api_key_file).read().strip()
|
|
# Store the API key in the environment so that our children
|
|
# don't need to read it in
|
|
os.environ["HUMBUG_API_KEY"] = api_key
|
|
|
|
humbug_account_email = options.user + "@mit.edu"
|
|
import humbug
|
|
humbug_client = humbug.Client(
|
|
email=humbug_account_email,
|
|
api_key=api_key,
|
|
verbose=True,
|
|
client="zephyr_mirror",
|
|
site=options.site)
|
|
|
|
start_time = time.time()
|
|
|
|
if options.sync_subscriptions:
|
|
configure_logger(logger, None) # make the output cleaner
|
|
logger.info("Syncing your ~/.zephyr.subs to your Humbug Subscriptions!")
|
|
add_humbug_subscriptions(True)
|
|
sys.exit(0)
|
|
|
|
# Kill all zephyr_mirror processes other than this one and its parent.
|
|
if not options.test_mode:
|
|
pgrep_query = "/usr/bin/python.*zephyr_mirror"
|
|
if options.shard is not None:
|
|
pgrep_query = "%s.*--shard=%s" % (pgrep_query, options.shard)
|
|
proc = subprocess.Popen(['pgrep', '-U', os.environ["USER"], "-f", pgrep_query],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE)
|
|
out, _err_unused = proc.communicate()
|
|
for pid in map(int, out.split()):
|
|
if pid == os.getpid() or pid == os.getppid():
|
|
continue
|
|
|
|
# Another copy of zephyr_mirror.py! Kill it.
|
|
logger.info("Killing duplicate zephyr_mirror process %s" % (pid,))
|
|
try:
|
|
os.kill(pid, signal.SIGINT)
|
|
except OSError:
|
|
# We don't care if the target process no longer exists, so just log the error
|
|
logger.exception("")
|
|
|
|
if options.shard is not None and set(options.shard) != set("a"):
|
|
# The shard that is all "a"s is the one that handles personals
|
|
# forwarding and humbug => zephyr forwarding
|
|
options.forward_personals = False
|
|
options.forward_from_humbug = False
|
|
|
|
if options.forward_from_humbug:
|
|
child_pid = os.fork()
|
|
if child_pid == 0:
|
|
CURRENT_STATE = States.HumbugToZephyr
|
|
# Run the humbug => zephyr mirror in the child
|
|
configure_logger(logger, "humbug=>zephyr")
|
|
zsig_fullname = fetch_fullname(options.user)
|
|
humbug_to_zephyr(options)
|
|
sys.exit(0)
|
|
else:
|
|
child_pid = None
|
|
CURRENT_STATE = States.ZephyrToHumbug
|
|
|
|
import zephyr
|
|
logger_name = "zephyr=>humbug"
|
|
if options.shard is not None:
|
|
logger_name += "(%s)" % (options.shard,)
|
|
configure_logger(logger, logger_name)
|
|
# Have the kernel reap children for when we fork off processes to send Humbugs
|
|
signal.signal(signal.SIGCHLD, signal.SIG_IGN)
|
|
zephyr_to_humbug(options)
|