#!/usr/bin/env python3 import bottle from bottle import get, post, redirect, request, response, view from config import config from db import db import logging import tweepy from sendmail import sendmail from session import SessionPlugin from mastodon import Mastodon def url(route): return '%s://%s/%s' % ( request.urlparts.scheme, request.urlparts.netloc, route) @get('/') @view('template/propaganda.tpl') def propaganda(): pass @post('/register') @view('template/register.tpl') def register_post(): try: email = request.forms['email'] password = request.forms['pass'] password_repeat = request.forms['pass-repeat'] city = request.forms['city'] except KeyError: return dict(error='Please, fill the form.') if password != password_repeat: return dict(error='Passwords do not match.') if db.by_email(email): return dict(error='Email address already in use.') # send confirmation mail try: link = url('confirm/' + city + '/%s' % db.user_token(email, password)) print(link) # only for local testing logger.error('confirmation link to ' + email + ": " + link) sendmail( email, "Confirm your account", body="Complete your registration here: %s" % (link) ) return dict(info='Confirmation mail sent.') except Exception: logger.error("Could not send confirmation mail to " + email, exc_info=True) return dict(error='Could not send confirmation mail.') @get('/confirm//') @view('template/propaganda.tpl') def confirm(city, token): # check whether city already exists if db.by_city(city): return dict(error='This Account was already confirmed, please try ' 'signing in.') # create db-entry if db.confirm(token, city): # :todo show info "Account creation successful." redirect('/settings') return dict(error='Account creation failed. Please try to register again.') @get('/version') def version(): import git repo = git.Repo(search_parent_directories=True) return repo.head.object.hexsha @post('/login') @view('template/login.tpl') def login_post(): # check login try: if db.by_email(request.forms['email']) \ .check_password(request.forms['pass']): redirect('/settings') except KeyError: return dict(error='Please, fill the form.') except AttributeError: pass return dict(error='Authentication failed.') @get('/city/') def city_page(city, info=None): citydict = db.user_facing_properties(city) if citydict is not None: citydict['info'] = info return bottle.template('template/city.tpl', **citydict) return bottle.template('template/propaganda.tpl', **dict(info='There is no Ticketfrei bot in your city' ' yet. Create one yourself!')) @get('/city/mail/') @view('template/mail.tpl') def display_mail_page(city): user = db.by_city(city) return user.state() @post('/city/mail/submit/') def subscribe_mail(city): email = request.forms['mailaddress'] token = db.mail_subscription_token(email, city) confirm_link = url('city/mail/confirm/' + token) print(confirm_link) # only for local testing # send mail with code to email sendmail(email, "Subscribe to Ticketfrei " + city + " Mail Notifications", body="To subscribe to the mail notifications for Ticketfrei " + city + ", click on this link: " + confirm_link, city=city) return city_page(city, info="Thanks! You will receive a confirmation mail.") @get('/city/mail/confirm/') def confirm_subscribe(token): email, city = db.confirm_subscription(token) user = db.by_city(city) user.add_subscriber(email) return city_page(city, info="Thanks for subscribing to mail notifications!") @get('/city/mail/unsubscribe/') def unsubscribe(token): email, city = db.confirm_subscription(token) user = db.by_city(city) user.remove_subscriber(email) return city_page(city, info="You successfully unsubscribed " + email + " from the mail notifications.") @get('/settings') @view('template/settings.tpl') def settings(user): return user.state() @post('/settings/markdown') @view('template/settings.tpl') def update_markdown(user): user.set_markdown(request.forms['markdown']) return user.state() @post('/settings/mail_md') @view('template/settings.tpl') def update_mail_md(user): user.set_mail_md(request.forms['mail_md']) return user.state() @post('/settings/goodlist') @view('template/settings.tpl') def update_trigger_patterns(user): user.set_trigger_words(request.forms['goodlist']) return user.state() @post('/settings/blocklist') @view('template/settings.tpl') def update_badwords(user): user.set_badwords(request.forms['blocklist']) return user.state() @post('/settings/telegram') def register_telegram(user): apikey = request.forms['apikey'] user.update_telegram_key(apikey) return city_page(user.get_city(), info="Thanks for registering Telegram!") # unused afaik #@get('/api/state') #def api_enable(user): # return user.state() @get('/static/') def static(filename): return bottle.static_file(filename, root='static') @get('/guides/') def guides(filename): return bottle.static_file(filename, root='guides') @get('/logout/') def logout(): # clear auth cookie response.set_cookie('uid', '', expires=0, path="/") response.set_cookie('csrf', '', expires=0, path="/") # :todo show info "Logout successful." response.set_cookie('csrf', '', expires=0, path="/") redirect('/') @get('/login/twitter') def login_twitter(user): """ Starts the twitter OAuth authentication process. :return: redirect to twitter. """ consumer_key = config["twitter"]["consumer_key"] consumer_secret = config["twitter"]["consumer_secret"] callback_url = url("login/twitter/callback") auth = tweepy.OAuthHandler(consumer_key, consumer_secret, callback_url) try: redirect_url = auth.get_authorization_url() except tweepy.TweepError: logger.error('Twitter OAuth Error: Failed to get request token.', exc_info=True) return dict(error="Failed to get request token.") user.save_request_token(auth.request_token) redirect(redirect_url) @get('/login/twitter/callback') def twitter_callback(user): """ Gets the callback :return: """ # twitter passes the verifier/oauth token secret in a GET request. verifier = request.query['oauth_verifier'] consumer_key = config["twitter"]["consumer_key"] consumer_secret = config["twitter"]["consumer_secret"] auth = tweepy.OAuthHandler(consumer_key, consumer_secret) request_token = user.get_request_token() auth.request_token = request_token auth.get_access_token(verifier) user.save_twitter_token(auth.access_token, auth.access_token_secret) redirect("/settings") @post('/login/mastodon') def login_mastodon(user): """ Mastodon OAuth authentication process. :return: redirect to city page. """ # get app tokens instance_url = request.forms.get('instance_url') masto_email = request.forms.get('email') masto_pass = request.forms.get('pass') client_id, client_secret = user.get_mastodon_app_keys(instance_url) m = Mastodon(client_id=client_id, client_secret=client_secret, api_base_url=instance_url) try: access_token = m.log_in(masto_email, masto_pass) user.save_masto_token(access_token, instance_url) return city_page(user.get_city(), info='Thanks for supporting decentralized social networks!') except Exception: logger.error('Login to Mastodon failed.', exc_info=True) return dict(error='Login to Mastodon failed.') logger = logging.getLogger() fh = logging.FileHandler('/var/log/ticketfrei/error.log') fh.setLevel(logging.DEBUG) logger.addHandler(fh) application = bottle.default_app() bottle.install(SessionPlugin('/')) if __name__ == '__main__': bottle.run(host="0.0.0.0", port=config["web"]["port"]) else: application.catchall = False