gear-orders/web/app.py

import sys
import requests
import json
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 markupsafe import Markup
from textwrap import dedent
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 web.api import api

DEBUG_VITE_PREFIX = '/v/'
PROD_VITE_PREFIX = '/static/'

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

try:
  with open("web/static/.vite/manifest.json") as f:
    vite_manifest = json.load(f)
except FileNotFoundError:
  vite_manifest = None

@app.context_processor
def inject_vite_entry():
  def vite_entry(file):
    if vite_manifest is None:
      return Markup(
        dedent(f"""
          <script type="module" src="{DEBUG_VITE_PREFIX}@vite/client"></script>
          <script type="module" src="{DEBUG_VITE_PREFIX}{file}"></script>
          """).strip())
    elif file in vite_manifest:
        src = vite_manifest[file]['file']
        result = f'<script type="module" src="{PROD_VITE_PREFIX}{src}"></script>'
        if('css' in vite_manifest[file]):
          for css in vite_manifest[file]['css']:
            result += f'<link href="{PROD_VITE_PREFIX}{css}" rel="stylesheet" />'
        return Markup(
          dedent(result.strip()))
  return dict(vite_entry=vite_entry)

@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:
    print('Redirect to dashboard')
    return redirect('/dashboard/')

@app.route('/profile/', defaults={'path': ''})
@app.route('/dashboard/', defaults={'path': ''})
@app.route('/orders/<path:path>')
@app.route('/dashboard/<path:path>')
@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] if key in data else ''))
  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:
    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:
    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)