# encoding: utf-8
import datetime
import urlparse
import werkzeug
import werkzeug.urls
import pvl.login.auth
import pvl.web
import pvl.web.response
from pvl.login import pubtkt
from pvl.web import urls
from pvl.web import html5 as html
import logging; log = logging.getLogger('pvl.login.server')
class Handler (pvl.web.Handler) :
# Bootstrap
DOCTYPE = 'html'
HTML_XMLNS = None
HTML_LANG = 'en'
CSS = (
'//netdna.bootstrapcdn.com/bootstrap/3.0.3/css/bootstrap.min.css',
)
JS = (
'//code.jquery.com/jquery.js',
'//netdna.bootstrapcdn.com/bootstrap/3.0.3/js/bootstrap.min.js',
)
STYLE = """
body {
padding-top: 2em;
text-align: center;
}
.container {
padding: 2em 1em;
text-align: left;
}
"""
def redirect (self, *url, **params) :
return pvl.web.response.redirect(self.url(*url, **params))
pubtkt = None
invalid_pubtkt = None
valid_pubtkt = None
def init (self) :
self.alerts = []
def alert (self, type, alert, icon=None) :
log.info(u"%s: %s", type, alert)
self.alerts.append((type, icon, unicode(alert)))
def process_cookie (self) :
"""
Reverse the urlencoding used for the cookie...
"""
log.debug("cookies: %s", self.request.cookies)
cookie = self.request.cookies.get(self.app.cookie_name)
if not cookie :
return
log.debug("cookie %s=%s", self.app.cookie_name, cookie)
cookie = werkzeug.urls.url_unquote(cookie)
log.debug("cookie decoded: %s", cookie)
if not cookie :
return
try :
self.pubtkt = self.app.load(cookie)
except pubtkt.ParseError as ex :
self.alert('danger', ex, icon='compare')
except pubtkt.VerifyError as ex :
self.alert('danger', ex, icon='warning-sign')
self.invalid_pubtkt = ex.pubtkt
except pubtkt.ExpiredError as ex :
self.alert('warning', ex, icon='clock')
# store it anyways, but not as valid
self.pubtkt = ex.pubtkt
else :
# it's a parsed, verified and valid pubtkt
self.valid_pubtkt = self.pubtkt
return self.pubtkt
def process_back (self) :
self.server = None
self.back = urlparse.urlunparse((self.app.login_scheme, self.app.login_server, '/', '', '', ''))
back = self.request.args.get('back')
if back :
url = urlparse.urlparse(back, self.app.login_scheme)
if not self.app.login_scheme :
scheme = url.scheme
elif url.scheme == self.app.login_scheme :
scheme = url.scheme
else :
self.alert('info', "Using SSL for application URL", icon='lock')
scheme = self.app.login_scheme
if url.hostname :
self.server = self.app.check_server(url.hostname)
else :
self.server = self.app.login_server
self.back = urlparse.urlunparse((scheme, self.server, url.path, url.params, url.query, url.fragment))
def render_alerts (self) :
for type, icon, alert in self.alerts :
yield html.div(class_='alert alert-{type}'.format(type=type))(
html.span(class_='glyphicon glyphicon-{glyphicon}'.format(glyphicon=icon)) if icon else None,
alert
)
class Index (Handler) :
TITLE = u"Päivölä Network Login"
STYLE = Handler.STYLE + """
.pubtkt {
width: 30em;
margin: 1em auto;
}
.pubtkt form {
display: inline;
}
.pubtkt .panel-heading {
line-height: 20px;
}
.pubtkt .panel-body .glyphicon {
width: 1em;
float: left;
line-height: 20px;
}
.pubtkt .panel-body .progress {
margin-bottom: 0;
margin-left: 2em;
}
"""
JS = Handler.JS + (
'/static/pubtkt-expire.js',
)
def process (self) :
self.process_cookie()
if not self.pubtkt :
return self.redirect(Login)
def render_valid (self, valid) :
seconds = valid.seconds + valid.days * (24 * 60 * 60)
minutes, seconds = divmod(seconds, 60)
hours, minutes = divmod(minutes, 60)
return "%2d:%02d:%02d" % (hours, minutes, seconds)
def render_status (self, pubtkt) :
valid = pubtkt.valid()
grace = pubtkt.grace()
if grace :
return 'warning'
elif valid :
return 'success'
else :
return 'danger'
def render_pubtkt_valid (self, pubtkt) :
"""
Yield HTML for ticket validity.
"""
lifetime = self.app.login_valid
valid = pubtkt.valid()
grace = pubtkt.grace()
grace_period = pubtkt.grace_period()
remaining = pubtkt.remaining()
if valid :
progress = float(valid.seconds) / float(lifetime.seconds)
else :
progress = None
if grace :
title = "Remaining renewal period"
label = "{grace} (Renew)".format(grace=self.render_valid(grace))
status = 'warning'
elif valid :
title = "Remaining validity period"
label = "{valid}".format(valid=self.render_valid(valid))
status = 'success'
else :
title = "Expired"
label = "Expired"
status = 'danger'
if progress :
return html.div(class_='panel-body', title=title)(
html.span(class_='glyphicon glyphicon-time'),
html.div(class_='progress pubtkt-progress',
data_start=valid.seconds,
data_refresh=grace_period.seconds if remaining else None,
data_end=lifetime.seconds,
)(
html.div(class_='progress-bar progress-bar-{status}'.format(status=status),
role='progressbar',
style='width: {pp:.0f}%'.format(pp=progress*100),
)(
html.span(class_='pubtkt-progress-label')(label)
)
)
)
else :
return None # html.p(label)
def render_pubtkt_fields (self, pubtkt) :
"""
Yield (glyphicon, text) to render as fields for ticket.
"""
if pubtkt.cip :
yield 'cloud', None, "Network address", pubtkt.cip
if pubtkt.udata :
yield 'comment', None, "User data", pubtkt.udata
for token in pubtkt.tokens :
yield 'flag', None, "Access token", token
if pubtkt.bauth :
yield 'keys', None, "Authentication token", pubtkt.bauth
def render_pubtkt (self, pubtkt) :
status = self.render_status(pubtkt)
domain = self.app.login_domain
return html.div(class_='pubtkt panel panel-{status}'.format(status=status))(
html.div(class_='panel-heading')(
html.span(class_='glyphicon glyphicon-user'),
html.strong(pubtkt.uid),
html.span("@", domain),
),
self.render_pubtkt_valid(pubtkt),
html.ul(class_='list-group')(
html.li(class_='list-group-item {status}'.format(status=('alert-'+status if status else '')), title=title)(
html.span(class_='glyphicon glyphicon-{glyphicon}'.format(glyphicon=icon)) if icon else None,
data,
) for icon, status, title, data in self.render_pubtkt_fields(pubtkt)
),
html.div(class_='panel-footer')(
#html.div(class_='btn-toolbar', role='toolbar')(
(
html.form(action=self.url(Login), method='post', class_='form-inline')(
html.button(type='submit', class_='btn btn-success')(
html.span(class_='glyphicon glyphicon-time'), "Renew"
)
)
) if pubtkt.valid() else (
html.form(action=self.url(Login), method='get', class_='form-inline')(
html.button(type='submit', class_='btn btn-info')(
html.span(class_='glyphicon glyphicon-log-in'), "Login"
)
),
),
html.form(action=self.url(Logout), method='post', class_='form-inline pull-right')(
html.button(type='submit', class_='btn btn-warning')(
html.span(class_='glyphicon glyphicon-log-out'), "Logout"
)
),
#),
),
)
def render (self) :
return html.div(class_='container')(
self.render_alerts(),
self.render_pubtkt(self.pubtkt) if self.pubtkt else None,
)
class Login (Handler) :
TITLE = "Login"
STYLE = Handler.STYLE + """
form#login {
max-width: 50%;
padding: 1em;
margin: 0 auto;
}
"""
login_failure = None
def process (self) :
self.process_cookie()
try :
self.process_back()
except pubtkt.Error as ex :
self.alert('danger', ex)
if self.pubtkt :
self.username = self.pubtkt.uid
else :
self.username = None
# update cookie?
set_pubtkt = None
if self.request.method == 'POST' :
username = self.request.form.get('username')
password = self.request.form.get('password')
if username :
# preprocess
username = username.strip().lower()
if username and password :
self.username = username
try :
set_pubtkt = self.app.auth(username, password)
except pvl.login.auth.AuthError as ex :
self.alert('danger', "Internal authentication error, try again later?")
else :
if not set_pubtkt :
self.alert('danger', "Invalid authentication credentials, try again.")
elif self.pubtkt and self.pubtkt.valid() :
# renew manually if valid
set_pubtkt = self.app.renew(self.pubtkt)
# a POST request that does not modify state is a failure
if not set_pubtkt :
self.login_failure = True
elif 'renew' in self.request.args :
# renew automatically if in grace period
if self.pubtkt and self.pubtkt.grace() :
set_pubtkt = self.app.renew(self.pubtkt)
if set_pubtkt :
signed = self.app.sign(set_pubtkt)
self.pubtkt = set_pubtkt
# browsers and mod_pubtkt seem to be very particular about quoting ;'s in cookie values...
# this follows PHP's setcookie() encoding, without any quoting of the value..
cookie = '{cookie}={value}; Domain={domain}; Secure; HttpOnly'.format(
cookie = self.app.cookie_name,
value = werkzeug.urls.url_quote(signed),
domain = self.app.cookie_domain,
)
# redirect with cookie
response = pvl.web.response.redirect(self.back)
response.headers.add('Set-Cookie', cookie)
return response
def status (self) :
if self.login_failure :
return 400
else :
return 200
def render (self) :
domain = self.app.login_domain
if 'logout' in self.request.args :
self.alert('info', "You have been logged out.", icon='log-out')
if self.pubtkt and self.pubtkt.valid() :
renew = True
# within validity period...
self.alert('info', "Login or renew ticket.", icon='log-in')
else :
renew = False
return html.div(class_='container')(
html.form(action=self.url(back=self.back), method='POST', id='login')(
self.render_alerts(),
html.fieldset(
html.legend(
(
"Login @ ",
html.a(href=self.back)(self.server),
) if self.server else (
"Login"
)
),
html.div(class_='form-group')(
html.div(class_='input-group')(
html.label(for_='username', class_='sr-only')("Username"),
html.input(name='username', type='text', class_='form-control', placeholder="username", required=True, autofocus=(not self.username), value=self.username),
html.span(class_='input-group-addon')("@{domain}".format(domain=domain)),
),
html.label(for_='password', class_='sr-only')("Password"),
html.input(name='password', type='password', class_='form-control', placeholder="Password", required=(not renew), autofocus=bool(self.username)),
),
html.button(type='submit', class_='btn btn-primary')(
html.span(class_='glyphicon glyphicon-log-in'), "Login"
),
html.button(type='submit', class_='btn btn-success')(
html.span(class_='glyphicon glyphicon-time'), "Renew"
) if renew else None,
)
)
)
class Logout (Handler) :
TITLE = "Logout"
def process (self) :
self.process_cookie()
if not self.pubtkt :
return self.redirect(Login)
if self.request.method == 'POST' :
response = pvl.web.response.redirect(self.url(Login, logout=1))
response.set_cookie(self.app.cookie_name, '',
expires = 0,
domain = self.app.cookie_domain,
secure = self.app.cookie_secure,
httponly = self.app.cookie_httponly,
)
return response
def render (self) :
return html.div(class_='container')(
html.form(action=self.url(), method='post')(
self.render_alerts(),
html.fieldset(
html.legend("Logout {pubtkt.uid}".format(pubtkt=self.pubtkt)),
html.button(type='submit', class_='btn btn-warning')(
html.span(class_='glyphicon glyphicon-log-out'), "Logout"
),
)
)
)
class SSL (Handler) :
TITLE = "SSL"
OUT = 'tmp/spkac'
def render_cert (self) :
return html.div(class_='container')(
self.render_alerts(),
html.div(class_='alert alert-success')(
"Your new SSL client cert has been signed, and should shortly be installed within your browser."
)
)
def respond_cert (self, cert) :
"""
Generate a response for a signed cert, showing the user an informational page, and redirecting to the cert itself..
"""
location = self.url(SSL, cert=cert)
return pvl.web.Response(
self.render_html(
body = self.render_cert(),
extrahead = html.meta(http_equiv='refresh', content='0;{location}'.format(location=location)),
),
status = 200,
#headers = {
# 'Location': location
#},
mimetype = 'text/html',
)
def process_spkac (self, spkac) :
log.info("SPKAC: %s", spkac)
try :
cert = self.app.ssl_sign(self.pubtkt, spkac)
except pvl.login.ssl.Error as ex :
self.alert('danger', ex)
return
log.info("Redirecting to client cert: %s", cert)
return self.respond_cert(cert)
def process_cert (self, cert) :
"""
Return user cert as download.
Uses the application/x-x509-user-cert mimetype per
https://developer.mozilla.org/en-US/docs/NSS_Certificate_Download_Specification
"""
try :
file = self.app.ssl_open(self.pubtkt, cert)
except pvl.login.ssl.Error as ex :
self.alert('danger', ex)
return
log.info("Returning client cert: %s", file)
return pvl.web.Response(self.response_file(file), mimetype='application/x-x509-user-cert')
def process (self, cert=None) :
if not self.process_cookie() :
return self.redirect(Login, back=self.url())
self.sslcert_dn = self.request.headers.get('X-Forwarded-SSL-DN')
if cert :
return self.process_cert(cert)
if self.request.method == 'POST' :
spkac = self.request.form.get('spkac')
if spkac:
return self.process_spkac(spkac)
def render (self) :
if self.sslcert_dn :
self.alert('info', "You are currently using a client SSL cert: {self.sslcert_dn}".format(self=self))
return html.div(class_='container')(
html.form(action=self.url(), method='post')(
self.render_alerts(),
html.fieldset(
html.legend("SSL Login"),
html.keygen(name='spkac', challenge='foo', keytype='RSA'),
html.button(type='submit', class_='btn')(
"Generate Certificate"
),
)
)
)
class LoginApplication (pvl.web.Application) :
URLS = urls.Map((
urls.rule('/', Index),
urls.rule('/login', Login),
urls.rule('/logout', Logout),
# proto
urls.rule('/ssl', SSL),
urls.rule('/ssl/<cert>', SSL),
))
PUBLIC_KEY = 'etc/login/public.pem'
PRIVATE_KEY = 'etc/login/private.pem'
login_domain = 'test.paivola.fi'
login_server = 'login.test.paivola.fi'
login_valid = datetime.timedelta(minutes=60)
login_grace = datetime.timedelta(minutes=15)
login_scheme = 'https'
cookie_name = 'auth_pubtkt'
cookie_domain = 'test.paivola.fi'
cookie_secure = True
cookie_httponly = True
def __init__ (self, auth, ssl=None, public_key=PUBLIC_KEY, private_key=PRIVATE_KEY, **opts) :
super(LoginApplication, self).__init__(**opts)
self._auth = auth
self._ssl = ssl
self.server_keys = pubtkt.ServerKeys.config(
public_key = public_key,
private_key = private_key,
)
def check_server (self, server) :
"""
Check that the given target server is valid.
"""
server = server.lower()
if server == self.login_domain or server.endswith('.' + self.login_domain) :
return server
else :
raise pubtkt.ServerError("Target server is not covered by our authentication domain: {domain}".format(domain=self.login_domain))
def load (self, cookie) :
"""
Load a pubtkt from a cookie, and verify it.
"""
return pubtkt.PubTkt.load(cookie, self.server_keys.public)
def auth (self, username, password) :
"""
Perform authentication, returning a PubTkt (unsiigned) or None.
Raises auth.AuthError.
"""
auth = self._auth.auth(username, password)
if not auth :
return None
return pubtkt.PubTkt.new(username,
valid = self.login_valid,
grace = self.login_grace,
tokens = list(self._auth.access(auth)),
udata = self._auth.userdata(auth),
)
def sign (self, pubtkt) :
"""
Create a cookie by signing the given pubtkt.
"""
return pubtkt.sign(self.server_keys.private)
def renew (self, pubtkt) :
"""
Renew and re-sign the given pubtkt.
"""
auth = self._auth.renew(pubtkt.uid)
if not auth :
raise pubtkt.RenewError("Unable to re-authenticate")
return pubtkt.update(
valid = self.login_valid,
grace = self.login_grace,
tokens = list(self._auth.access(auth)),
udata = self._auth.userdata(auth),
)
def ssl_sign (self, pubtkt, spkac) :
"""
Generate a SSL client cert for the given user.
Returns the redirect token for downloading it.
Raises pvl.login.ssl.Error
"""
if not self._ssl :
raise pvl.login.ssl.Error("No ssl CA available for signing")
return self._ssl.sign_user(pubtkt.uid, spkac,
userinfo = pubtkt.udata,
)
def ssl_open (self, pubtkt, cert) :
"""
Open and return an SSL cert file.
Raises pvl.login.ssl.Error
"""
return self._ssl.open_cert(pubtkt.uid, cert)