Files
zulip/zerver/views/zephyr.py
Tim Abbott b064559652 zephyr: Add strict assertion about username format.
This ensures that even if it were possible to create an MIT Kerberos
account with a malicious username and/or hack webathena to pretend
that's the case, one couldn't do anything malicious.

This security improvement only impacts a single installation of Zulip
where Zephyr mirroring is in use that has already had the fix applied,
so there's no reason to do a security notice for it.

Found by Graham Bleaney using pysa.
2020-03-17 05:37:25 -07:00

62 lines
2.3 KiB
Python

from django.conf import settings
from django.http import HttpResponse, HttpRequest
from django.utils.translation import ugettext as _
from zerver.decorator import authenticated_json_view
from zerver.lib.ccache import make_ccache
from zerver.lib.request import has_request_variables, REQ
from zerver.lib.response import json_success, json_error
from zerver.lib.users import get_api_key
from zerver.models import UserProfile
import base64
import logging
import re
import subprocess
import ujson
from typing import Optional
# Hack for mit.edu users whose Kerberos usernames don't match what they zephyr
# as. The key is for Kerberos and the value is for zephyr.
kerberos_alter_egos = {
'golem': 'ctl',
}
@authenticated_json_view
@has_request_variables
def webathena_kerberos_login(request: HttpRequest, user_profile: UserProfile,
cred: Optional[str]=REQ(default=None)) -> HttpResponse:
global kerberos_alter_egos
if cred is None:
return json_error(_("Could not find Kerberos credential"))
if not user_profile.realm.webathena_enabled:
return json_error(_("Webathena login not enabled"))
try:
parsed_cred = ujson.loads(cred)
user = parsed_cred["cname"]["nameString"][0]
if user in kerberos_alter_egos:
user = kerberos_alter_egos[user]
assert(user == user_profile.email.split("@")[0])
# Limit characters in usernames to valid MIT usernames
# This is important for security since DNS is not secure.
assert(re.match(r'^[a-z0-9_.-]+$', user) is not None)
ccache = make_ccache(parsed_cred)
except Exception:
return json_error(_("Invalid Kerberos cache"))
# TODO: Send these data via (say) rabbitmq
try:
api_key = get_api_key(user_profile)
subprocess.check_call(["ssh", settings.PERSONAL_ZMIRROR_SERVER, "--",
"/home/zulip/python-zulip-api/zulip/integrations/zephyr/process_ccache",
user,
api_key,
base64.b64encode(ccache).decode("utf-8")])
except Exception:
logging.exception("Error updating the user's ccache")
return json_error(_("We were unable to setup mirroring for you"))
return json_success()