diff --git a/.env_sample b/.env_sample index bcea691..95b0f9b 100644 --- a/.env_sample +++ b/.env_sample @@ -1,3 +1,9 @@ ENVIRONMENT="Development" FLASK_SECRET_KEY="156a7fbb77c17708e41f044ea5131be2ad592a7deb21866bcb63d7801d2fa13e" FLASK_DATABASE=compy.sqlite +# Password for the admin interface (http://localhost:5000/admin). +# For deployments, remove FLASK_ADMIN_PASSWORD and set FLASK_ADMIN_PASSWORD_HASH +# instead. Generate a hash with: +# python3 -c "from werkzeug.security import generate_password_hash; import getpass; print(generate_password_hash(getpass.getpass()))" +FLASK_ADMIN_PASSWORD="compy-admin" +#FLASK_ADMIN_PASSWORD_HASH="" diff --git a/Readme.md b/Readme.md index 9c1fff1..41ee898 100644 --- a/Readme.md +++ b/Readme.md @@ -57,8 +57,12 @@ Install Weasyprint https://doc.courtbouillon.org/weasyprint/stable/first_steps.h - Execute `git clone https://github.com/Azrael3000/Compy.git` - Switch to the new folder: `cd Compy` - Set up the environmen: `cp .env_sample .env` - - For deployments you MUST edit the .env file and provide a new secret. A new one can be generated e.g. by running - `python3 -c "import secrets; print(secrets.token_hex())"` + - For deployments you MUST edit the .env file: + - Provide a new secret (used to sign the admin session cookie). A new one can be generated e.g. by running + `python3 -c "import secrets; print(secrets.token_hex())"` + - Set your own admin password. Either change `FLASK_ADMIN_PASSWORD`, or (recommended) remove it and set + `FLASK_ADMIN_PASSWORD_HASH` to a password hash generated by running + `python3 -c "from werkzeug.security import generate_password_hash; import getpass; print(generate_password_hash(getpass.getpass()))"` - Start a virtual environment and install required packages: - Linux: `source venv/bin/activate && pip install -r requirements.txt` - Set up the database and run the server: `python3 compy.py --init_db` @@ -73,8 +77,9 @@ Install Weasyprint https://doc.courtbouillon.org/weasyprint/stable/first_steps.h - Linux: `python3 compy.py` - Windows: `python3.exe compy.py` - Navigate your browser to `localhost:5000` - - The admin interface is at `localhost:5000/admin?auth=XXXXXX` where `XXXXXX` are the first 6 - characters of your `FLASK_SECRET_KEY` from `.env` + - The admin interface is at `localhost:5000/admin`. It asks for the admin password configured in + `.env` (`FLASK_ADMIN_PASSWORD` or `FLASK_ADMIN_PASSWORD_HASH`); a login is valid for 12 hours + or until you press "Logout" ## Test data diff --git a/compy_concurrency_test.py b/compy_concurrency_test.py index ad894fa..bb84286 100644 --- a/compy_concurrency_test.py +++ b/compy_concurrency_test.py @@ -21,7 +21,7 @@ class TestConcurrentPages(compy_testing.CompyServerTestCase): @classmethod def setUpClass(cls): super().setUpClass() - session = requests.Session() + session = cls.adminSession() # name the default competition and upload the excel file response = session.post(cls.base_url + "/competition", @@ -119,18 +119,20 @@ def registrationWrites(self, session, round_index): def testConcurrentPagesDoNotInterfere(self): page_simulations = [ - ("admin1", self.adminTabCompOne), - ("admin2", self.adminTabCompTwo), - ("clock", self.clockDisplay), - ("judge", self.judgePhone), - ("results", self.publicResultsPage), - ("registration", self.registrationWrites), + ("admin1", self.adminTabCompOne, True), + ("admin2", self.adminTabCompTwo, True), + ("clock", self.clockDisplay, False), + ("judge", self.judgePhone, False), + ("results", self.publicResultsPage, False), + ("registration", self.registrationWrites, True), ] failures = [] stop_event = threading.Event() - def run_page(page_name, request_round): - page_session = requests.Session() + def run_page(page_name, request_round, is_admin_page): + # admin pages carry a session cookie, public pages must work + # without any authentication + page_session = self.adminSession() if is_admin_page else requests.Session() for round_index in range(N_ROUNDS): if stop_event.is_set(): return @@ -151,7 +153,7 @@ def run_page(page_name, request_round): self.assertEqual(failures, []) # after the storm: comp 1 must be fully intact - session = requests.Session() + session = self.adminSession() response = session.post(self.base_url + "/load_comp", json={"comp_id": self.comp_one_id}) self.assertEqual(response.json()["comp_name"], "Comp One") self.assertEqual(len(response.json()["athletes"]), 30) @@ -175,9 +177,57 @@ def testForgedJudgeHashIsRejected(self): "block": self.first_block, "lane": "1"}) self.assertEqual(response.status_code, 404) # ...and it must not have switched or broken anything + response = self.adminSession().get(self.base_url + "/athletes", + params={"comp_id": self.comp_one_id}) + self.assertEqual(len(response.json()["athletes"]), 30) + + def testAdminEndpointsRequireLogin(self): + # without a session cookie all admin endpoints must refuse to act response = requests.get(self.base_url + "/athletes", params={"comp_id": self.comp_one_id}) - self.assertEqual(len(response.json()["athletes"]), 30) + self.assertEqual(response.status_code, 401) + response = requests.post(self.base_url + "/competition", + json={"comp_name": "Hacked", "overwrite": True, + "comp_id": self.comp_one_id}) + self.assertEqual(response.status_code, 401) + response = requests.delete(self.base_url + "/competition", + json={"comp_id": self.comp_one_id}) + self.assertEqual(response.status_code, 401) + # the admin page itself redirects to the login form + response = requests.get(self.base_url + "/admin", allow_redirects=False) + self.assertEqual(response.status_code, 302) + self.assertTrue(response.headers["Location"].endswith("/admin/login")) + # ...and nothing was changed by the rejected requests + response = self.adminSession().post(self.base_url + "/load_comp", + json={"comp_id": self.comp_one_id}) + self.assertEqual(response.json()["comp_name"], "Comp One") + + def testWrongPasswordIsRejected(self): + session = requests.Session() + response = session.post(self.base_url + "/admin/login", + data={"password": "not-the-password"}) + self.assertEqual(response.status_code, 401) + response = session.get(self.base_url + "/athletes", + params={"comp_id": self.comp_one_id}) + self.assertEqual(response.status_code, 401) + + def testJudgeCanSaveResultWithoutAdminSession(self): + # a judge phone is not logged in as admin; the judge hash from the + # QR code must be enough to save a result, a forged hash must not be + response = requests.get(self.base_url + "/judge/athletes", + params={"comp_id": self.comp_one_id, "judge_id": self.judge_id, + "judge_hash": self.judge_hash, "day": self.first_day, + "block": self.first_block, "lane": "1"}) + start_id = response.json()["lane_list"][0]["s_id"] + result = {"comp_id": self.comp_one_id, "judge_id": self.judge_id, + "id": start_id, "rp": "", "penalty": 0, "card": "WHITE", + "remarks": "", "judge_remarks": ""} + response = requests.put(self.base_url + "/result", + json=result | {"judge_hash": "deadbeef"}) + self.assertEqual(response.status_code, 401) + response = requests.put(self.base_url + "/result", + json=result | {"judge_hash": self.judge_hash}) + self.assertEqual(response.status_code, 200) if __name__ == '__main__': diff --git a/compy_flask.py b/compy_flask.py index d79f548..df8de86 100644 --- a/compy_flask.py +++ b/compy_flask.py @@ -24,11 +24,16 @@ # # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ +import hmac import logging +import time +from datetime import timedelta +from functools import wraps from compy_data import CompyData from compy_config import CompyConfig -from flask import Flask, render_template, request, send_file, Response, make_response, current_app +from flask import Flask, render_template, request, send_file, Response, make_response, session, redirect, url_for from os import path, mkdir +from werkzeug.security import check_password_hash from werkzeug.utils import secure_filename from werkzeug.routing import IntegerConverter try: @@ -58,23 +63,63 @@ def __init__(self, app, db, start_flask): app.config['UPLOAD_FOLDER'] = self.config_.upload_folder app.url_map.converters['signed_int'] = self.SignedIntConverter + # admin sessions are stored in a cookie signed with SECRET_KEY; + # the cookie is not readable by page javascript and not sent on + # cross-site requests (basic CSRF protection) + app.config['SESSION_COOKIE_HTTPONLY'] = True + app.config['SESSION_COOKIE_SAMESITE'] = 'Lax' + # one login lasts a full competition day + app.permanent_session_lifetime = timedelta(hours=12) + + if not app.config.get('ADMIN_PASSWORD_HASH') and not app.config.get('ADMIN_PASSWORD'): + logging.error("Neither FLASK_ADMIN_PASSWORD_HASH nor FLASK_ADMIN_PASSWORD is set " + "in the .env file; logging in to the admin interface is not possible") + + def admin_required(f): + """Only allow the request if this browser has an admin session. + + The admin page itself redirects to the login form, all other + (api) endpoints return 401 so the frontend can react. + """ + @wraps(f) + def wrapper(*args, **kwargs): + if session.get('is_admin'): + return f(*args, **kwargs) + if request.method == 'GET' and request.path == '/admin': + return redirect(url_for('login')) + return self.unauthorized() + return wrapper + @app.route('/admin', methods=['GET']) + @admin_required def admin(): return self.admin() + @app.route('/admin/login', methods=['GET', 'POST']) + def login(): + return self.login() + + @app.route('/admin/logout', methods=['GET']) + def logout(): + return self.logout() + @app.route('/upload_file', methods=['POST']) + @admin_required def uploadFile(): return self.uploadFile() @app.route('/store_results', methods=['POST']) + @admin_required def storeResults(): return self.storeResults() @app.route('/upload_sponsor_img', methods=['POST']) + @admin_required def uploadSponsorImg(): return self.uploadSponsorImg() @app.route('/competition', methods=['POST', 'DELETE']) + @admin_required def changeCompName(): if request.method == 'POST': return self.changeCompName() @@ -82,18 +127,22 @@ def changeCompName(): return self.deleteComp() @app.route('/change_special_ranking_name', methods=['POST']) + @admin_required def changeSpecialRankingName(): return self.changeSpecialRankingName() @app.route('/change_registration', methods=['POST']) + @admin_required def changeRegistration(): return self.changeRegistration() @app.route('/load_comp', methods=['POST']) + @admin_required def loadComp(): return self.loadComp() @app.route('/start_list', methods=['GET', 'PUT']) + @admin_required def startList(): if request.method == 'GET': return self.startList() @@ -101,45 +150,59 @@ def startList(): return self.updateStartList() @app.route('/start_list_pdf', methods=['GET']) + @admin_required def startListPDF(): return self.startListPDF() @app.route('/breaks', methods=['GET']) + @admin_required def breaks(): return self.breaks() @app.route('/lane_list', methods=['GET']) + @admin_required def laneList(): return self.laneList() @app.route('/lane_list_pdf', methods=['GET']) + @admin_required def laneListPDF(): return self.laneListPDF() @app.route('/result', methods=['GET', 'PUT']) def result(): if request.method == 'GET': + # admin page only + if not session.get('is_admin'): + return self.unauthorized() return self.result(False) elif request.method == 'PUT': + # used by the admin page and by judge phones; + # updateResult checks the admin session or the judge hash return self.updateResult() @app.route('/result_pdf', methods=['GET']) + @admin_required def resultPDF(): return self.result(True) @app.route('/change_lane_style', methods=['POST']) + @admin_required def changeLaneStyle(): return self.changeLaneStyle() @app.route('/change_comp_type', methods=['POST']) + @admin_required def changeCompType(): return self.changeCompType() @app.route('/change_selected_country', methods=['POST']) + @admin_required def changeSelectedCountry(): return self.changeSelectedCountry() @app.route('/judge', methods=['DELETE', 'POST']) + @admin_required def judge(): if request.method == 'DELETE': return self.deleteJudge() @@ -147,14 +210,17 @@ def judge(): return self.addJudge() @app.route('/judge/qr_code', methods=['GET']) + @admin_required def judgeQrCode(): return self.getJudgeQrCode() @app.route('/judges', methods=['GET']) + @admin_required def judges(): return self.getJudges() @app.route('/athlete', methods=['DELETE', 'POST']) + @admin_required def athlete(): if request.method == 'DELETE': return self.deleteAthlete() @@ -162,10 +228,12 @@ def athlete(): return self.addAthlete() @app.route('/athletes', methods=['GET']) + @admin_required def athletes(): return self.getAthletes() @app.route('/national_records', methods=['GET']) + @admin_required def nationalRecords(): return self.nationalRecords() @@ -187,10 +255,12 @@ def judgeAthleteResult(): return self.getJudgeAthleteResult() @app.route('/disciplines/', methods=['GET']) + @admin_required def disciplines(federation): return self.disciplines(federation) @app.route('/block', methods=['POST', 'UPDATE', 'DELETE']) + @admin_required def block(): if request.method == 'POST': return self.modifyBlock(True) @@ -204,6 +274,7 @@ def clock(comp_id, current, offset): return self.getClock(comp_id, current, offset) @app.route('/publish_results', methods=['UPDATE']) + @admin_required def publish_results(): return self.updatePublishResults() @@ -609,6 +680,10 @@ def updateResult(self): comp = self.getData(request) if comp is None: return self.badRequest("Failed to load competition") + # results are entered by the admin page (session cookie) or by a + # judge phone (judge id + hash in the request body) + if not session.get('is_admin') and not self.isValidJudge(request, comp): + return self.unauthorized() content, status = self.handleRequest(request, ['id', 'rp', 'penalty', 'card', 'remarks', 'judge_remarks'], CompyData.updateResult, comp) if status != 200: logging.debug("Failed to set result") @@ -926,11 +1001,49 @@ def getClock(self, comp_id, current, offset): "offset": offset} return render_template('clock.html', **content) + def checkAdminPassword(self, password): + """Compare a login attempt against the configured admin password. + + FLASK_ADMIN_PASSWORD_HASH (a werkzeug password hash, recommended + for deployments) takes precedence over the plain text + FLASK_ADMIN_PASSWORD. Both comparisons are constant time. If + neither is configured, logging in is not possible. + """ + pw_hash = self.app_.config.get('ADMIN_PASSWORD_HASH') + if pw_hash: + return check_password_hash(pw_hash, password) + pw = self.app_.config.get('ADMIN_PASSWORD') + if pw: + return hmac.compare_digest(pw.encode('utf-8'), password.encode('utf-8')) + return False + + def login(self): + if session.get('is_admin'): + return redirect(url_for('admin')) + error = None + if request.method == 'POST': + password = request.form.get('password', '') + if self.checkAdminPassword(password): + session.clear() + session['is_admin'] = True + session.permanent = True + logging.info("Admin login from " + str(request.remote_addr)) + return redirect(url_for('admin')) + # throttle brute force attempts + time.sleep(1) + logging.warning("Failed admin login attempt from " + str(request.remote_addr)) + error = "Wrong password" + content = {"version": self.version(), "error": error} + return make_response(render_template('login.html', **content), 401 if error else 200) + + def logout(self): + session.clear() + return redirect(url_for('login')) + + def unauthorized(self): + return {"status": "error", "error_msg": "Authentication required"}, 401 + def admin(self): - auth = request.args.get('auth') - if auth != current_app.config["SECRET_KEY"][:6]: - content = {"version": self.version()} - return make_response(render_template('404.html', **content), 404) all_countries = country_converter.CountryConverter().data["IOC"].dropna().to_list() # the admin frontend loads competition 1 after the page is ready, so # pre-fill the name field with that competition diff --git a/compy_testing.py b/compy_testing.py index a1edf61..0420844 100644 --- a/compy_testing.py +++ b/compy_testing.py @@ -19,6 +19,7 @@ import unittest import flask +import requests from werkzeug.serving import make_server import compy_data @@ -27,12 +28,14 @@ REPO_ROOT = os.path.dirname(os.path.abspath(__file__)) TEST_COMPETITION_XLSX = os.path.join(REPO_ROOT, "test_competition.xlsx") +ADMIN_PASSWORD = "compy-test-password" def makeApp(database_path): app = flask.Flask("compy", root_path=REPO_ROOT) app.config["DATABASE"] = database_path app.config["SECRET_KEY"] = "0123456789abcdef0123456789abcdef_compy_test" + app.config["ADMIN_PASSWORD"] = ADMIN_PASSWORD return app @@ -96,3 +99,13 @@ def tearDownClass(cls): cls.server.shutdown() cls.server_thread.join() super().tearDownClass() + + @classmethod + def adminSession(cls): + """A requests session that is logged in to the admin interface.""" + session = requests.Session() + response = session.post(cls.base_url + "/admin/login", + data={"password": ADMIN_PASSWORD}) + if response.status_code != 200 or not session.cookies: + raise AssertionError("admin login failed in test setup") + return session diff --git a/static/compy.js b/static/compy.js index 029f6af..92705a1 100644 --- a/static/compy.js +++ b/static/compy.js @@ -25,6 +25,13 @@ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ */ +// if the admin session has expired (or is missing) every api call returns +// 401; send the user to the login page in that case +$(document).ajaxError(function(event, jqxhr) { + if (jqxhr.status == 401) + window.location.href = "/admin/login"; +}); + var _global_prev_name = ""; var _days_with_disciplines_lanes = null; var _comp_id = null; diff --git a/templates/template.html b/templates/template.html index e66d478..c3392b1 100644 --- a/templates/template.html +++ b/templates/template.html @@ -56,6 +56,7 @@

Compy {{version}}

Lane lists Results Clock + Logout
diff --git a/tests/compy.resource b/tests/compy.resource index 529a331..53ecefe 100644 --- a/tests/compy.resource +++ b/tests/compy.resource @@ -10,8 +10,9 @@ ${URL} localhost ${PORT} 5000 ${PATH} . ${BASE_URL} http://${URL}:${PORT}/${PATH} -${ADMIN_KEY} 156a7f# -${ADMIN_URL} ${BASE_URL}/admin?auth=${ADMIN_KEY} +# must match FLASK_ADMIN_PASSWORD in the .env used by the server under test +${ADMIN_PASSWORD} compy-admin +${ADMIN_URL} ${BASE_URL}/admin # Elements ${SETTINGS_BUTTON} id=settings_button >> a ${COMP_NAME_FIELD} id=comp_name @@ -23,6 +24,11 @@ Open Admin Page # Debug enable # Open Browser New Page ${ADMIN_URL} + ${login_form} = Get Element Count id=admin_password + IF ${login_form} > 0 + Fill Text id=admin_password ${ADMIN_PASSWORD} + Click id=admin_login_button + END Get Text h1 contains Compy Goto Settings