Got mgmt panel authenticating via indieauth

ownchatbot/web_panels.py

@@ -9,17 +9,61 @@ import json
 import emoji
 from ownchatbot.kofi_handlers import save_kofi_settings, kofi_pngs
 import random
+import pkce
+import requests
+from functools import wraps
 
 ocb = Blueprint('web_panels', __name__)
 
+state_value = ''
+
+
+def requires_login(f):
+    @wraps(f)
+    def decorated_function(*args, **kwargs):
+        if 'user' not in session:
+            return redirect(url_for('web_panels.login'))
+        return f(*args, **kwargs)
+    return decorated_function
+
+
+@ocb.route('/login')
+def login():  # Verify the streamer using indieauth, to their owncast instance
+    code_verifier, code_challenge = pkce.generate_pkce_pair()  # Generate a code verifier and code challenge
+    global state_value
+    state_value = code_verifier
+    owncast_url = current_app.config['OWNCAST_URL']
+    client_id = current_app.config['INDIEAUTH_CLIENT_ID']
+    redirect_url = f'{owncast_url}/api/auth/provider/indieauth?client_id={client_id}&redirect_uri={url_for("web_panels.auth_response", _external=True)}&response_type=code&code_challenge_method=S256&code_challenge={code_challenge}&state={code_verifier}'
+    return redirect(redirect_url)
+
+
+@ocb.route('/auth_response')
+def auth_response():
+    code = request.args.get('code')
+    state = request.args.get('state')
+    if state == state_value:  # Check that the state value returned matches the state value sent
+        current_app.logger.info(f'Valid CSRF Code. Streamer authenticated.')
+
+        user_info = 'code'
+
+        session['user'] = user_info
+        return redirect(url_for('web_panels.mgmt'))
+    else:
+        current_app.logger.info(f'Invalid CSRF Code. Streamer not authenticated.')
+        return 'Not Authorized'
+
+
+@ocb.route('/logout')
+def logout():
+    session.pop('user', None)
+    return redirect(url_for('web_panels.user_panel'))
+
 
 @ocb.route('/mgmt', methods=['GET'])  # The streamer's management panel
+@requires_login
 def mgmt():
-    auth_code = request.args.get('auth')
-    if auth_code == current_app.config['MGMT_AUTH']:
-        session['auth_code'] = auth_code  # Store auth code in session
-    else:
-        return "Not authorized", 403
+    owncast_url = current_app.config['OWNCAST_URL']
     db = get_db()
     users = get_all_users(db)
     utc_timezone = timezone.utc
@@ -36,7 +80,6 @@ def mgmt():
     gunicorn_logging = current_app.config['GUNICORN']
     prefix = current_app.config['PREFIX']
     access_token = current_app.config['ACCESS_TOKEN']
-    owncast_url = current_app.config['OWNCAST_URL']
     kofi_token = current_app.config['KOFI_TOKEN']
     kofi_integration = current_app.config['KOFI_INTEGRATION']
     kofi_logos = kofi_pngs()
@@ -106,15 +149,17 @@ def user_panel():
 
 
 @ocb.route('/mgmt/fulfill', methods=['GET'])
+@requires_login
 def fulfilled():
     db = get_db()
     reward_id = request.args.get('reward_id')
     username = request.args.get('username')
     fulfill_reward(db, reward_id)
-    return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+    return redirect(url_for('web_panels.mgmt'))
 
 
 @ocb.route('/mgmt/refund', methods=['GET'])
+@requires_login
 def refund():
     db = get_db()
     reward_id = request.args.get('reward_id')
@@ -125,13 +170,12 @@ def refund():
     user_id = request.args.get('rewarder_id')
     refund_points(db, user_id, points)  # resets points
     refund_reward(db, reward_id)  # marks the reward as refunded
-    return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+    return redirect(url_for('web_panels.mgmt'))
 
 
 @ocb.route('/mgmt/edit_account/<user_id>', methods=['GET', 'POST'])  # Streamer manually edit user's account
+@requires_login
 def edit_account(user_id):
-    if 'auth_code' not in session:
-        return "Not authorized", 403
     db = get_db()
     name = request.args.get('name')
     points = request.args.get('points')
@@ -150,7 +194,7 @@ def edit_account(user_id):
                     current_app.logger.info(f'Removed {name}\'s email')
                 else:
                     current_app.logger.info(f'Changed {name}\'s email to {newemail}')
-        return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+        return redirect(url_for('web_panels.mgmt'))
 
     return render_template('edit_account.html',
                            name=name,
@@ -160,6 +204,7 @@ def edit_account(user_id):
 
 
 @ocb.route('/mgmt/delete/<reward_name>', methods=['GET', 'POST'])
+@requires_login
 def delete(reward_name):
     del_reward = current_app.config['REWARDS']
     del_reward.pop(reward_name)
@@ -169,13 +214,12 @@ def delete(reward_name):
             if reread_votes():
                 if reread_goals():
                     pass
-    return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+    return redirect(url_for('web_panels.mgmt'))
 
 
 @ocb.route('/mgmt/edit/<reward_name>', methods=['GET', 'POST'])
+@requires_login
 def edit(reward_name):
-    if 'auth_code' not in session:
-        return "Not authorized", 403
     active_categories = current_app.config['ACTIVE_CAT']
     all_the_rewards = current_app.config['REWARDS']
     reward_data = all_the_rewards[reward_name]
@@ -219,7 +263,7 @@ def edit(reward_name):
             reread_goals()
         if reward_data['type'] == 'vote':
             reread_votes()
-        return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+        return redirect(url_for('web_panels.mgmt'))
     
     return render_template('edit.html',
                            all_cats=all_cats,
@@ -229,6 +273,7 @@ def edit(reward_name):
 
 
 @ocb.route('/mgmt/settings', methods=['GET', 'POST'])  # OwnchatBot settings panel
+@requires_login
 def settings():
     points_interval = int(request.form['points_interval'])
     points_award = int(request.form['points_award'])
@@ -253,10 +298,11 @@ def settings():
     if save_config(config_dict):  # Save new config.py
         current_app.logger.info('Saved new config.')
         
-    return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+    return redirect(url_for('web_panels.mgmt'))
 
 
 @ocb.route('/mgmt/announcements', methods=['GET', 'POST'])  # OwnchatBot settings panel
+@requires_login
 def announcements():
     announce_enable = 'announce_enable' in request.form
     announce_interval = int(request.form['announce_interval'])
@@ -270,10 +316,11 @@ def announcements():
     if save_announce(announce_dict):  # Save new announce.py
         current_app.logger.info('Saved new announcements.')
         
-    return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+    return redirect(url_for('web_panels.mgmt'))
 
 
 @ocb.route('/mgmt/ksettings', methods=['GET', 'POST'])  # OwnchatBot settings panel
+@requires_login
 def ksettings():
     kofi_settings_dict = current_app.config['KOFI_SETTINGS']
     if request.method == 'POST':
@@ -292,13 +339,12 @@ def ksettings():
         if save_kofi_settings(kofi_settings_dict):
             current_app.logger.info(f'Saved Kofi settings')
     
-    return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+    return redirect(url_for('web_panels.mgmt'))
 
 
 @ocb.route('/mgmt/add/<reward_type>', methods=['GET', 'POST'])
+@requires_login
 def add(reward_type):
-    if 'auth_code' not in session:
-        return "Not authorized", 403
     all_cats = current_app.config['ALL_CAT']
     active_categories = current_app.config['ACTIVE_CAT']
     all_the_rewards = current_app.config['REWARDS']
@@ -376,7 +422,7 @@ def add(reward_type):
             inactive_categories = current_app.config['INACTIVE_CAT']
             inactive_categories.append(name)  # Add it to the INACTIVE_CAT variable
             reread_categories()  # Write it to categories.py
-        return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+        return redirect(url_for('web_panels.mgmt'))
 
     return render_template('add.html',
                            all_cats=all_cats,
@@ -408,13 +454,13 @@ def set_viewer_email():
 @ocb.route('/mgmt/activate/<category>', methods=['GET', 'POST'])
 def activate(category):
     activate_category(category)
-    return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+    return redirect(url_for('web_panels.mgmt'))
 
 
 @ocb.route('/mgmt/deactivate/<category>', methods=['GET', 'POST'])
 def deactivate(category):
     deactivate_category(category)
-    return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+    return redirect(url_for('web_panels.mgmt'))
 
 
 @ocb.route('/mgmt/delcat/<cat_name>/<cat_act>', methods=['GET', 'POST'])
@@ -431,7 +477,7 @@ def delcat(cat_name, cat_act):
         if cat_name in details['categories']:
             details['categories'].remove(cat_name)
     save_rewards(current_rewards)
-    return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+    return redirect(url_for('web_panels.mgmt'))
 
 
 @ocb.route('/mgmt/reset/<reward_name>/<reward_type>', methods=['GET', 'POST'])  # Reset votes and goals to zero
@@ -440,22 +486,22 @@ def reset(reward_name, reward_type):
         reset_goal(reward_name)
     if reward_type == "vote":
         reset_vote(reward_name)
-    return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+    return redirect(url_for('web_panels.mgmt'))
 
 
 @ocb.route('/mgmt/rereadvotes', methods=['GET', 'POST'])
 def rereadv():
     reread_votes()
-    return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+    return redirect(url_for('web_panels.mgmt'))
 
 
 @ocb.route('/mgmt/clearfulfilled', methods=['GET', 'POST'])
 def clearfulfilled():
     clear_fulfilled_rewards()
-    return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+    return redirect(url_for('web_panels.mgmt'))
 
 
 @ocb.route('/mgmt/clearqueue', methods=['GET', 'POST'])
 def clear_queue():
     clear_reward_queue()
-    return redirect(url_for('web_panels.mgmt', auth=session['auth_code']))
+    return redirect(url_for('web_panels.mgmt'))