import sys import requests from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from flask import Flask, render_template, request, redirect, flash, abort from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user from flask_wtf.csrf import CSRFProtect import hashlib import hmac from settings import FLASK_SECRET_KEY, TELEGRAM_API_TOKEN, TELEGRAM_BOT_NAME, TELEGRAM_BOT_DOMAIN, MASTODON_OAUTH_REDIRECT_URI from db.queries import user_get, user_mastodon_user_set from api import api app = Flask(__name__) app.secret_key = FLASK_SECRET_KEY login_manager = LoginManager() login_manager.init_app(app) login_manager.login_view = "index" csrf = CSRFProtect() csrf.init_app(app) class FlaskUser(UserMixin): def __init__(self, db_user): self.db_user = db_user def get_id(self): return self.db_user.telegram_username @login_manager.user_loader def load_user(user_id): try: db_user = user_get(user_id) return FlaskUser(db_user) except: return None app.register_blueprint(api, url_prefix='/api') @app.route('/') def index(): if not current_user.is_authenticated: data = {'bot_name': TELEGRAM_BOT_NAME, 'bot_damin': TELEGRAM_BOT_DOMAIN} return render_template('index.html', data = data) else: return redirect('/dashboard/') @app.route('/dashboard/', defaults={'path': ''}) @app.route('/orders/') @app.route('/profile', defaults={'path': ''}) @app.route('/dashboard/') @login_required def dashboard(path): return render_template('dashboard.html') @app.route('/logout') def logout(): logout_user() return redirect('/') def string_generator(data_incoming): data = data_incoming.copy() del data['hash'] keys = sorted(data.keys()) string_arr = [] for key in keys: string_arr.append(key + '=' + data[key]) string_cat = '\n'.join(string_arr) return string_cat @app.route('/oauth') @login_required def mastodon_oauth(): code = request.args['code'] server = current_user.db_user.mastodon_server payload = { 'grant_type': 'authorization_code', 'client_id': server.client_id, 'client_secret': server.client_secret, 'redirect_uri': MASTODON_OAUTH_REDIRECT_URI, 'code': code } token_request = requests.post(f"https://{server.name}/oauth/token", data=payload) if not token_request.ok: import pdb; pdb.set_trace() abort(500) access_token = token_request.json()['access_token'] headers = { "Authorization": f"Bearer {access_token}" } verify_request = requests.get(f"https://{server.name}/api/v1/accounts/verify_credentials", headers=headers) if not verify_request.ok: import pdb; pdb.set_trace() abort(500) username = verify_request.json()['username'] user_mastodon_user_set(current_user.db_user.id, username, access_token) return redirect('/profile') @app.route('/login') def login(): tg_data = { "id": request.args.get("id", None), "first_name": request.args.get('first_name', None), "last_name": request.args.get('last_name', None), "username": request.args.get("username", None), "photo_url": request.args.get("photo_url", None), "auth_date": request.args.get('auth_date', None), "hash": request.args.get("hash", None) } data_check_string = string_generator(tg_data) secret_key = hashlib.sha256(TELEGRAM_API_TOKEN.encode('utf-8')).digest() secret_key_bytes = secret_key data_check_string_bytes = bytes(data_check_string, 'utf-8') hmac_string = hmac.new(secret_key_bytes, data_check_string_bytes, hashlib.sha256).hexdigest() if hmac_string == tg_data['hash']: try: db_user = user_get(tg_data['username']) db_user.telegram_photo_url = request.args.get("photo_url") db_user.save() login_user(FlaskUser(db_user)) except: flash("Login failed. Please try again.") return redirect('/') else: flash("Login failed. Please try again.") return redirect('/') return redirect('/dashboard/') if __name__ == '__main__': app.run(host='0.0.0.0', debug=True, port=8080)