775 lines
24 KiB
Python
775 lines
24 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import secrets
|
|
from datetime import datetime
|
|
from functools import wraps
|
|
|
|
import bcrypt
|
|
import config
|
|
from flask import (
|
|
Flask,
|
|
abort,
|
|
flash,
|
|
jsonify,
|
|
make_response,
|
|
redirect,
|
|
render_template,
|
|
request,
|
|
send_from_directory,
|
|
session,
|
|
)
|
|
from flask_limiter import Limiter
|
|
from flask_limiter.util import get_remote_address
|
|
from flask_session import Session
|
|
from flask_wtf.csrf import CSRFError, CSRFProtect, generate_csrf
|
|
from pymongo import MongoClient
|
|
from pymongo.errors import DuplicateKeyError, PyMongoError
|
|
from redis import Redis
|
|
|
|
|
|
APP_ROOT = pathlib.Path(__file__).resolve().parent
|
|
SONGS_DIR = pathlib.Path(
|
|
os.environ.get('TAIKO_WEB_SONGS_DIR', APP_ROOT / 'public' / 'songs')
|
|
).resolve()
|
|
EXPERIENCE_SETTINGS_ID = 'single-song'
|
|
EXPERIENCE_DIFFICULTIES = ('easy', 'normal', 'hard', 'oni', 'ura')
|
|
EASTER_EGG_CODE_ALPHABET = 'ABCDEFGHJKLMNPQRSTUVWXYZ23456789'
|
|
EASTER_EGG_RESULT_FIELDS = (
|
|
'points',
|
|
'good',
|
|
'ok',
|
|
'bad',
|
|
'max_combo',
|
|
'drumroll',
|
|
'gauge',
|
|
)
|
|
|
|
|
|
def take_config(name, default=None, required=False):
|
|
if hasattr(config, name):
|
|
return getattr(config, name)
|
|
if required:
|
|
raise ValueError('Missing config option: {}'.format(name))
|
|
return default
|
|
|
|
|
|
def load_or_create_secret_key():
|
|
configured = os.environ.get('TAIKO_WEB_SECRET_KEY') or take_config('SECRET_KEY')
|
|
if isinstance(configured, str) and configured != 'change-me' and len(configured) >= 32:
|
|
return configured
|
|
|
|
secret_path = pathlib.Path(
|
|
os.environ.get('TAIKO_WEB_SECRET_KEY_FILE', APP_ROOT / '.taiko-secret-key')
|
|
).resolve()
|
|
secret_path.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
descriptor = os.open(secret_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
except FileExistsError:
|
|
pass
|
|
else:
|
|
with os.fdopen(descriptor, 'w', encoding='ascii') as output:
|
|
output.write(secrets.token_hex(32))
|
|
output.flush()
|
|
os.fsync(output.fileno())
|
|
secret = secret_path.read_text(encoding='ascii').strip()
|
|
if len(secret) < 32:
|
|
raise RuntimeError('TAIKO_WEB_SECRET_KEY must contain at least 32 characters.')
|
|
return secret
|
|
|
|
|
|
basedir = take_config('BASEDIR', '/') or '/'
|
|
if not basedir.startswith('/'):
|
|
basedir = '/' + basedir
|
|
if not basedir.endswith('/'):
|
|
basedir += '/'
|
|
|
|
app = Flask(__name__, static_folder=None)
|
|
app.secret_key = load_or_create_secret_key()
|
|
app.config['WTF_CSRF_CHECK_DEFAULT'] = False
|
|
app.jinja_env.globals.setdefault('csrf_token', generate_csrf)
|
|
|
|
redis_config = dict(take_config('REDIS', {}, required=True))
|
|
redis_host = os.environ.get('TAIKO_WEB_REDIS_HOST') or redis_config.get(
|
|
'CACHE_REDIS_HOST', '127.0.0.1'
|
|
)
|
|
redis_client = Redis(
|
|
host=redis_host,
|
|
port=redis_config.get('CACHE_REDIS_PORT', 6379),
|
|
password=redis_config.get('CACHE_REDIS_PASSWORD'),
|
|
db=redis_config.get('CACHE_REDIS_DB') or 0,
|
|
socket_connect_timeout=1,
|
|
socket_timeout=1,
|
|
)
|
|
try:
|
|
redis_client.ping()
|
|
app.config.update(
|
|
SESSION_TYPE='redis',
|
|
SESSION_REDIS=redis_client,
|
|
)
|
|
except Exception:
|
|
app.config.update(
|
|
SESSION_TYPE='filesystem',
|
|
SESSION_FILE_DIR=str(APP_ROOT / 'flask_session'),
|
|
)
|
|
Session(app)
|
|
csrf = CSRFProtect(app)
|
|
limiter = Limiter(
|
|
get_remote_address,
|
|
app=app,
|
|
storage_uri=os.environ.get('REDIS_URI') or 'memory://',
|
|
in_memory_fallback_enabled=True,
|
|
)
|
|
|
|
mongo_config = take_config('MONGO', required=True)
|
|
client = MongoClient(
|
|
host=os.environ.get('TAIKO_WEB_MONGO_HOST') or mongo_config['host']
|
|
)
|
|
db = client[mongo_config['database']]
|
|
|
|
|
|
def ensure_indexes():
|
|
indexes = (
|
|
(db.users, 'username', {'unique': True}),
|
|
(db.users, 'username_lower', {'unique': True}),
|
|
(db.songs, 'id', {'unique': True}),
|
|
(db.easter_egg_codes, 'code', {'unique': True}),
|
|
(db.easter_egg_codes, [('created_at', -1)], {}),
|
|
(db.easter_egg_codes, [('song_id', 1), ('created_at', -1)], {}),
|
|
)
|
|
for collection, keys, options in indexes:
|
|
try:
|
|
collection.create_index(keys, **options)
|
|
except PyMongoError:
|
|
app.logger.exception('Unable to create index on %s', collection.name)
|
|
|
|
|
|
ensure_indexes()
|
|
|
|
|
|
def site_path(path=''):
|
|
return basedir + path.lstrip('/')
|
|
|
|
|
|
def safe_int(value, default=None):
|
|
if value in (None, ''):
|
|
return default
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def safe_float(value, default=None):
|
|
if value in (None, ''):
|
|
return default
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def route_song_id(value):
|
|
value = str(value)
|
|
return int(value) if re.fullmatch(r'\d+', value) else value
|
|
|
|
|
|
def safe_lang_map(value):
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def normalize_courses(value):
|
|
value = value if isinstance(value, dict) else {}
|
|
output = {}
|
|
for difficulty in EXPERIENCE_DIFFICULTIES:
|
|
course = value.get(difficulty)
|
|
output[difficulty] = {
|
|
'stars': safe_int(course.get('stars'), 0),
|
|
'branch': bool(course.get('branch')),
|
|
} if isinstance(course, dict) else None
|
|
return output
|
|
|
|
|
|
def serialize_song(song):
|
|
if not song or song.get('id') is None:
|
|
return None
|
|
output = {
|
|
'id': song.get('id'),
|
|
'title': song.get('title') or 'Untitled',
|
|
'subtitle': song.get('subtitle') or '',
|
|
'title_lang': safe_lang_map(song.get('title_lang')),
|
|
'subtitle_lang': safe_lang_map(song.get('subtitle_lang')),
|
|
'courses': normalize_courses(song.get('courses')),
|
|
'type': song.get('type') if song.get('type') in ('tja', 'osu') else 'tja',
|
|
'music_type': song.get('music_type') or 'mp3',
|
|
'offset': safe_float(song.get('offset'), 0),
|
|
'preview': 0,
|
|
'volume': safe_float(song.get('volume'), 1),
|
|
'hash': song.get('hash') or song.get('title') or str(song.get('id')),
|
|
'category_id': song.get('category_id'),
|
|
'lyrics': False,
|
|
'song_type': song.get('song_type') or '',
|
|
}
|
|
category = db.categories.find_one({'id': output['category_id']})
|
|
output['category'] = category.get('title') if category else None
|
|
skin = db.song_skins.find_one({'id': song.get('skin_id')}) if song.get('skin_id') else None
|
|
output['song_skin'] = {
|
|
key: value
|
|
for key, value in (skin or {}).items()
|
|
if key not in ('_id', 'id')
|
|
}
|
|
return output
|
|
|
|
|
|
def normalize_admin_song(song):
|
|
output = serialize_song(song)
|
|
if not output:
|
|
return None
|
|
output['enabled'] = bool(song.get('enabled'))
|
|
return output
|
|
|
|
|
|
def get_version():
|
|
output = {
|
|
'commit': None,
|
|
'commit_short': '',
|
|
'version': None,
|
|
'url': take_config('URL', ''),
|
|
}
|
|
version_path = APP_ROOT / 'version.json'
|
|
if version_path.is_file():
|
|
try:
|
|
stored = json.loads(version_path.read_text(encoding='utf-8'))
|
|
except (OSError, ValueError):
|
|
stored = {}
|
|
for key in output:
|
|
if stored.get(key):
|
|
output[key] = stored[key]
|
|
return output
|
|
|
|
|
|
def absolute_asset_url(filename):
|
|
origin = (
|
|
os.environ.get('TAIKO_WEB_SITE_URL')
|
|
or take_config('SITE_URL')
|
|
or request.host_url.rstrip('/')
|
|
)
|
|
return origin.rstrip('/') + site_path('assets/' + filename)
|
|
|
|
|
|
def index_seo():
|
|
canonical = request.host_url.rstrip('/') + basedir
|
|
return {
|
|
'lang': 'cn',
|
|
'html_lang': 'zh-Hans',
|
|
'title': 'taiko.su | taiko.asia 彩蛋站点',
|
|
'heading': 'taiko.su 单曲活动',
|
|
'description': 'taiko.asia 的彩蛋站点。完成管理员指定的歌曲即可领取彩蛋码。',
|
|
'keywords': 'taiko.su, taiko.asia, 太鼓, 音乐游戏, 彩蛋码',
|
|
'disclaimer': '非官方粉丝制作的网页音乐游戏。',
|
|
'image_url': absolute_asset_url('img/favicon-512.png'),
|
|
'canonical_url': canonical,
|
|
'default_url': canonical,
|
|
'og_locale': 'zh_CN',
|
|
'alternate_urls': [],
|
|
}
|
|
|
|
|
|
def get_config():
|
|
songs_baseurl = take_config('SONGS_BASEURL', '/songs/', required=True)
|
|
assets_baseurl = take_config('ASSETS_BASEURL', '/assets/', required=True)
|
|
if not re.match(r'^https?://', songs_baseurl) and not songs_baseurl.startswith('/'):
|
|
songs_baseurl = site_path(songs_baseurl)
|
|
if not re.match(r'^https?://', assets_baseurl) and not assets_baseurl.startswith('/'):
|
|
assets_baseurl = site_path(assets_baseurl)
|
|
return {
|
|
'basedir': basedir,
|
|
'songs_baseurl': songs_baseurl,
|
|
'assets_baseurl': assets_baseurl,
|
|
'_version': get_version(),
|
|
}
|
|
|
|
|
|
def get_user_level(user):
|
|
return safe_int((user or {}).get('user_level'), 0)
|
|
|
|
|
|
def check_password(user, password):
|
|
try:
|
|
return bool(user) and bcrypt.checkpw(password, user.get('password', b''))
|
|
except (TypeError, ValueError):
|
|
return False
|
|
|
|
|
|
def ensure_user_session_id(user):
|
|
session_id = user.get('session_id') if user else None
|
|
if session_id:
|
|
return session_id
|
|
session_id = secrets.token_hex(24)
|
|
db.users.update_one({'_id': user['_id']}, {'$set': {'session_id': session_id}})
|
|
return session_id
|
|
|
|
|
|
def current_admin(minimum_level=50):
|
|
username = session.get('username')
|
|
if not username:
|
|
return None
|
|
user = db.users.find_one({'username': username})
|
|
return user if get_user_level(user) >= minimum_level else None
|
|
|
|
|
|
def admin_required(level=50):
|
|
def decorator(function):
|
|
@wraps(function)
|
|
def wrapper(*args, **kwargs):
|
|
if not current_admin(level):
|
|
return abort(403)
|
|
return function(*args, **kwargs)
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
def get_experience_settings():
|
|
return db.experience_settings.find_one({'_id': EXPERIENCE_SETTINGS_ID}) or {
|
|
'_id': EXPERIENCE_SETTINGS_ID
|
|
}
|
|
|
|
|
|
def available_difficulties(song):
|
|
courses = normalize_courses((song or {}).get('courses'))
|
|
return [
|
|
difficulty
|
|
for difficulty in EXPERIENCE_DIFFICULTIES
|
|
if courses.get(difficulty)
|
|
]
|
|
|
|
|
|
def resolve_difficulty(song, requested=None):
|
|
available = available_difficulties(song)
|
|
if requested in available:
|
|
return requested
|
|
for preferred in ('oni', 'hard', 'normal', 'easy', 'ura'):
|
|
if preferred in available:
|
|
return preferred
|
|
return None
|
|
|
|
|
|
def build_experience_payload(issue_ticket=False):
|
|
settings_doc = get_experience_settings()
|
|
song_id = settings_doc.get('song_id')
|
|
song = db.songs.find_one({'id': song_id, 'enabled': True}) if song_id is not None else None
|
|
difficulty = resolve_difficulty(song, settings_doc.get('difficulty'))
|
|
public_song = serialize_song(song) if song and difficulty else None
|
|
output = {
|
|
'status': 'ok',
|
|
'configured': bool(public_song and difficulty),
|
|
'song': public_song,
|
|
'difficulty': difficulty,
|
|
}
|
|
if not output['configured']:
|
|
output['message'] = '管理员尚未配置可游玩的歌曲。'
|
|
return output
|
|
if issue_ticket:
|
|
ticket = secrets.token_urlsafe(32)
|
|
session.pop('experience_claim', None)
|
|
session['experience_ticket'] = ticket
|
|
session['experience_song_id'] = public_song['id']
|
|
session['experience_difficulty'] = difficulty
|
|
output['ticket'] = ticket
|
|
return output
|
|
|
|
|
|
def hash_verification_token(token):
|
|
return hashlib.sha256(token.encode('utf-8')).hexdigest()
|
|
|
|
|
|
def verification_token_matches(token, settings_doc):
|
|
expected = settings_doc.get('verification_token_hash')
|
|
return bool(
|
|
isinstance(token, str)
|
|
and token
|
|
and isinstance(expected, str)
|
|
and len(expected) == 64
|
|
and secrets.compare_digest(hash_verification_token(token), expected)
|
|
)
|
|
|
|
|
|
def request_verification_token():
|
|
header = request.headers.get('Authorization', '').strip()
|
|
if header.lower().startswith('bearer '):
|
|
return header[7:].strip()
|
|
return request.headers.get('X-Verification-Token', '').strip()
|
|
|
|
|
|
def normalize_easter_egg_code(value):
|
|
value = str(value or '').strip().upper()
|
|
return value if re.fullmatch(
|
|
r'AI4-[A-HJ-NP-Z2-9]{4}-[A-HJ-NP-Z2-9]{4}', value
|
|
) else None
|
|
|
|
|
|
def sanitize_result(value):
|
|
value = value if isinstance(value, dict) else {}
|
|
output = {}
|
|
for field in EASTER_EGG_RESULT_FIELDS:
|
|
try:
|
|
number = int(float(value.get(field)))
|
|
except (TypeError, ValueError, OverflowError):
|
|
number = 0
|
|
output[field] = max(0, min(number, 1000000000))
|
|
return output
|
|
|
|
|
|
def create_easter_egg_code(song, difficulty, result):
|
|
for _attempt in range(32):
|
|
left = ''.join(secrets.choice(EASTER_EGG_CODE_ALPHABET) for _ in range(4))
|
|
right = ''.join(secrets.choice(EASTER_EGG_CODE_ALPHABET) for _ in range(4))
|
|
document = {
|
|
'code': 'AI4-{}-{}'.format(left, right),
|
|
'song_id': song['id'],
|
|
'song_hash': song.get('hash') or song.get('title'),
|
|
'song_title': song.get('title') or 'Untitled',
|
|
'difficulty': difficulty,
|
|
'result': sanitize_result(result),
|
|
'created_at': datetime.utcnow(),
|
|
'verified_count': 0,
|
|
}
|
|
try:
|
|
db.easter_egg_codes.insert_one(document)
|
|
return document
|
|
except DuplicateKeyError:
|
|
continue
|
|
raise RuntimeError('Unable to allocate an easter egg code.')
|
|
|
|
|
|
@app.before_request
|
|
def protect_admin_writes_and_session():
|
|
endpoint = request.endpoint or ''
|
|
if (
|
|
request.method in ('POST', 'PUT', 'PATCH', 'DELETE')
|
|
and (endpoint == 'admin_login' or endpoint.startswith('admin_'))
|
|
):
|
|
csrf.protect()
|
|
username = session.get('username')
|
|
session_id = session.get('session_id')
|
|
if username and session_id and not db.users.find_one({
|
|
'username': username,
|
|
'session_id': session_id,
|
|
}, {'_id': True}):
|
|
session.clear()
|
|
|
|
|
|
@app.after_request
|
|
def secure_headers(response):
|
|
response.headers.setdefault('X-Content-Type-Options', 'nosniff')
|
|
response.headers.setdefault('Referrer-Policy', 'strict-origin-when-cross-origin')
|
|
if request.path.startswith(site_path('admin')) or request.path == site_path('1128admin1128'):
|
|
response.headers['Cache-Control'] = 'private, no-store, max-age=0'
|
|
return response
|
|
|
|
|
|
@app.errorhandler(CSRFError)
|
|
def csrf_error(_error):
|
|
return jsonify({'status': 'error', 'message': 'invalid_csrf'}), 400
|
|
|
|
|
|
@app.errorhandler(429)
|
|
def rate_limit_error(_error):
|
|
return jsonify({'success': False, 'error': 'rate_limited'}), 429
|
|
|
|
|
|
@app.get(basedir)
|
|
def index():
|
|
return render_template(
|
|
'index.html',
|
|
config=get_config(),
|
|
version=get_version(),
|
|
seo=index_seo(),
|
|
)
|
|
|
|
|
|
@app.route(site_path('1128admin1128'), methods=['GET', 'POST'])
|
|
@limiter.limit('10 per minute', methods=['POST'])
|
|
def admin_login():
|
|
if request.method == 'GET' and current_admin():
|
|
return redirect(site_path('admin/experience'))
|
|
username = ''
|
|
if request.method == 'POST':
|
|
username = (request.form.get('username') or '').strip()
|
|
password = (request.form.get('password') or '').encode('utf-8')
|
|
user = db.users.find_one({'username_lower': username.lower()})
|
|
if get_user_level(user) >= 50 and check_password(user, password):
|
|
session.clear()
|
|
session['username'] = user['username']
|
|
session['session_id'] = ensure_user_session_id(user)
|
|
session.permanent = True
|
|
db.users.update_one(
|
|
{'_id': user['_id']},
|
|
{'$set': {'last_login_at': datetime.utcnow()}},
|
|
)
|
|
return redirect(site_path('admin/experience'))
|
|
flash('管理员账号或密码错误。', 'error')
|
|
return render_template(
|
|
'admin_login.html',
|
|
config=get_config(),
|
|
username=username,
|
|
)
|
|
|
|
|
|
@app.get(site_path('admin'))
|
|
@app.get(site_path('admin/overview'))
|
|
@admin_required()
|
|
def admin_root():
|
|
return redirect(site_path('admin/experience'))
|
|
|
|
|
|
@app.route(site_path('admin/experience'), methods=['GET', 'POST'])
|
|
@admin_required()
|
|
def admin_experience():
|
|
admin = current_admin()
|
|
if request.method == 'POST':
|
|
song_id = (request.form.get('song_id') or '').strip()
|
|
difficulty = (request.form.get('difficulty') or '').strip().lower()
|
|
verification_token = request.form.get('verification_token') or ''
|
|
clear_token = bool(request.form.get('clear_token'))
|
|
updates = {
|
|
'updated_at': datetime.utcnow(),
|
|
'updated_by': admin['username'],
|
|
}
|
|
unsets = {}
|
|
if song_id:
|
|
song = db.songs.find_one({
|
|
'id': route_song_id(song_id),
|
|
'enabled': True,
|
|
})
|
|
if not song:
|
|
flash('请选择一首已启用的歌曲。', 'error')
|
|
return redirect(site_path('admin/experience'))
|
|
if difficulty not in available_difficulties(song):
|
|
flash('所选歌曲不包含这个难度。', 'error')
|
|
return redirect(site_path('admin/experience'))
|
|
updates.update(song_id=song['id'], difficulty=difficulty)
|
|
else:
|
|
unsets.update(song_id='', difficulty='')
|
|
if clear_token:
|
|
unsets.update(
|
|
verification_token_hash='',
|
|
verification_token_updated_at='',
|
|
)
|
|
elif verification_token:
|
|
if not 16 <= len(verification_token) <= 512:
|
|
flash('验证 Token 长度必须在 16 到 512 个字符之间。', 'error')
|
|
return redirect(site_path('admin/experience'))
|
|
updates.update(
|
|
verification_token_hash=hash_verification_token(verification_token),
|
|
verification_token_updated_at=datetime.utcnow(),
|
|
)
|
|
update = {'$set': updates}
|
|
if unsets:
|
|
update['$unset'] = unsets
|
|
db.experience_settings.update_one(
|
|
{'_id': EXPERIENCE_SETTINGS_ID},
|
|
update,
|
|
upsert=True,
|
|
)
|
|
flash('单曲活动设置已保存。')
|
|
return redirect(site_path('admin/experience'))
|
|
|
|
settings_doc = get_experience_settings()
|
|
songs = [
|
|
normalize_admin_song(song)
|
|
for song in db.songs.find({'enabled': True}).sort([('title', 1), ('id', 1)])
|
|
]
|
|
codes = list(db.easter_egg_codes.find({}).sort('created_at', -1).limit(200))
|
|
return render_template(
|
|
'admin_experience.html',
|
|
admin=admin,
|
|
config=get_config(),
|
|
settings_doc=settings_doc,
|
|
songs=songs,
|
|
codes=codes,
|
|
code_count=db.easter_egg_codes.count_documents({}),
|
|
token_configured=bool(settings_doc.get('verification_token_hash')),
|
|
)
|
|
|
|
|
|
@app.post(site_path('admin/logout'))
|
|
@admin_required()
|
|
def admin_logout():
|
|
session.clear()
|
|
return redirect(site_path('1128admin1128'))
|
|
|
|
|
|
@app.get(site_path('api/config'))
|
|
def api_config():
|
|
return jsonify(get_config())
|
|
|
|
|
|
@app.get(site_path('api/categories'))
|
|
def api_categories():
|
|
categories = []
|
|
for category in db.categories.find({}, {'_id': False}):
|
|
item = dict(category)
|
|
categories.append(item)
|
|
return jsonify(categories)
|
|
|
|
|
|
@app.get(site_path('api/experience'))
|
|
@limiter.limit('60 per minute')
|
|
def api_experience():
|
|
response = jsonify(build_experience_payload(issue_ticket=True))
|
|
response.headers['Cache-Control'] = 'private, no-store, max-age=0'
|
|
return response
|
|
|
|
|
|
@app.post(site_path('api/easter-eggs/claim'))
|
|
@limiter.limit('30 per minute')
|
|
def api_easter_egg_claim():
|
|
payload = request.get_json(silent=True) or {}
|
|
ticket = payload.get('ticket')
|
|
if not isinstance(ticket, str) or not ticket:
|
|
return jsonify({'status': 'error', 'message': 'invalid_ticket'}), 400
|
|
ticket_hash = hashlib.sha256(ticket.encode('utf-8')).hexdigest()
|
|
previous = session.get('experience_claim') or {}
|
|
if (
|
|
previous.get('ticket_hash') == ticket_hash
|
|
and normalize_easter_egg_code(previous.get('code'))
|
|
):
|
|
return jsonify(status='ok', success=True, code=previous['code'])
|
|
expected = session.get('experience_ticket')
|
|
if not isinstance(expected, str) or not secrets.compare_digest(ticket, expected):
|
|
return jsonify({'status': 'error', 'message': 'invalid_ticket'}), 400
|
|
|
|
settings_doc = get_experience_settings()
|
|
song = db.songs.find_one({
|
|
'id': settings_doc.get('song_id'),
|
|
'enabled': True,
|
|
})
|
|
difficulty = resolve_difficulty(song, settings_doc.get('difficulty'))
|
|
if (
|
|
not song
|
|
or not difficulty
|
|
or song['id'] != session.get('experience_song_id')
|
|
or difficulty != session.get('experience_difficulty')
|
|
):
|
|
return jsonify({'status': 'error', 'message': 'experience_changed'}), 409
|
|
try:
|
|
document = create_easter_egg_code(song, difficulty, payload.get('result'))
|
|
except (PyMongoError, RuntimeError):
|
|
app.logger.exception('Unable to create an easter egg code')
|
|
return jsonify({'status': 'error', 'message': 'claim_failed'}), 500
|
|
|
|
for key in (
|
|
'experience_ticket',
|
|
'experience_song_id',
|
|
'experience_difficulty',
|
|
):
|
|
session.pop(key, None)
|
|
session['experience_claim'] = {
|
|
'ticket_hash': ticket_hash,
|
|
'code': document['code'],
|
|
}
|
|
return jsonify(status='ok', success=True, code=document['code'])
|
|
|
|
|
|
@app.route(site_path('api/easter-eggs/verify'), methods=['POST', 'OPTIONS'])
|
|
@limiter.limit('120 per minute')
|
|
def api_easter_egg_verify():
|
|
def response(payload, status=200):
|
|
output = jsonify(payload)
|
|
output.status_code = status
|
|
output.headers['Access-Control-Allow-Origin'] = '*'
|
|
output.headers['Access-Control-Allow-Methods'] = 'POST, OPTIONS'
|
|
output.headers['Access-Control-Allow-Headers'] = (
|
|
'Authorization, Content-Type, X-Verification-Token'
|
|
)
|
|
output.headers['Cache-Control'] = 'no-store'
|
|
return output
|
|
|
|
if request.method == 'OPTIONS':
|
|
return response({}, 204)
|
|
settings_doc = get_experience_settings()
|
|
if not settings_doc.get('verification_token_hash'):
|
|
return response({
|
|
'success': False,
|
|
'error': 'verification_not_configured',
|
|
}, 503)
|
|
if not verification_token_matches(request_verification_token(), settings_doc):
|
|
return response({
|
|
'success': False,
|
|
'error': 'invalid_credentials',
|
|
}, 401)
|
|
payload = request.get_json(silent=True) or {}
|
|
code = normalize_easter_egg_code(payload.get('code'))
|
|
document = db.easter_egg_codes.find_one({'code': code}) if code else None
|
|
if not document:
|
|
return response({'success': False})
|
|
db.easter_egg_codes.update_one(
|
|
{'_id': document['_id']},
|
|
{
|
|
'$inc': {'verified_count': 1},
|
|
'$set': {'last_verified_at': datetime.utcnow()},
|
|
},
|
|
)
|
|
created_at = document.get('created_at')
|
|
return response({
|
|
'success': True,
|
|
'code': document['code'],
|
|
'song_id': document.get('song_id'),
|
|
'song_title': document.get('song_title'),
|
|
'difficulty': document.get('difficulty'),
|
|
'claimed_at': (
|
|
created_at.isoformat(timespec='seconds') + 'Z'
|
|
if isinstance(created_at, datetime)
|
|
else None
|
|
),
|
|
})
|
|
|
|
|
|
def cache_static(response, seconds):
|
|
output = make_response(response)
|
|
output.headers['Cache-Control'] = 'public, max-age={}'.format(seconds)
|
|
return output
|
|
|
|
|
|
@app.get(site_path('src/<path:ref>'))
|
|
def source_file(ref):
|
|
return cache_static(send_from_directory(APP_ROOT / 'public' / 'src', ref), 3600)
|
|
|
|
|
|
@app.get(site_path('assets/<path:ref>'))
|
|
def asset_file(ref):
|
|
return cache_static(send_from_directory(APP_ROOT / 'public' / 'assets', ref), 3600)
|
|
|
|
|
|
@app.get(site_path('songs/<path:ref>'))
|
|
def song_file(ref):
|
|
return cache_static(send_from_directory(SONGS_DIR, ref), 604800)
|
|
|
|
|
|
@app.get(site_path('manifest.json'))
|
|
def manifest_file():
|
|
return cache_static(
|
|
send_from_directory(APP_ROOT / 'public', 'manifest.json'),
|
|
3600,
|
|
)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description='Run the single-song Taiko event.')
|
|
parser.add_argument('port', type=int, nargs='?', default=34801)
|
|
parser.add_argument('-b', '--bind-address', default='localhost')
|
|
parser.add_argument('-d', '--debug', action='store_true')
|
|
arguments = parser.parse_args()
|
|
app.run(
|
|
host=arguments.bind_address,
|
|
port=arguments.port,
|
|
debug=arguments.debug,
|
|
)
|