131 lines
3.6 KiB
Python
131 lines
3.6 KiB
Python
"""Web endpoints."""
|
|
|
|
import os
|
|
import logging
|
|
import random
|
|
from typing import Dict, List
|
|
from urllib.parse import urlparse
|
|
|
|
from flask import Flask, request, redirect, send_file
|
|
import jwt
|
|
import requests
|
|
|
|
logger = logging.getLogger(__name__)
|
|
app = Flask(__name__)
|
|
|
|
|
|
@app.route("/redirect/user")
|
|
def redirect_from_query():
|
|
"""Redirect to the URL provided in the next parameter."""
|
|
next_url = request.args.get("next", "/")
|
|
return redirect(next_url)
|
|
|
|
|
|
@app.route("/redirect/validated")
|
|
def redirect_validated():
|
|
"""Redirect to the next parameter after a netloc check."""
|
|
next_url = request.args.get("next", "/")
|
|
parsed = urlparse(next_url)
|
|
if parsed.netloc and parsed.netloc != "example.com":
|
|
return redirect("/")
|
|
return redirect(next_url)
|
|
|
|
|
|
@app.route("/redirect/relative_only")
|
|
def redirect_relative():
|
|
"""Redirect to the next parameter after a scheme check."""
|
|
next_url = request.args.get("next", "/")
|
|
if "://" in next_url:
|
|
return redirect("/")
|
|
return redirect(next_url)
|
|
|
|
|
|
@app.route("/files/download")
|
|
def download_file():
|
|
"""Send the file at the supplied filename relative to the file root."""
|
|
filename = request.args.get("file", "readme.txt")
|
|
filepath = os.path.join("/var/www/files", filename)
|
|
return send_file(filepath)
|
|
|
|
|
|
@app.route("/files/realpath_download")
|
|
def download_file_realpath():
|
|
"""Send a file after resolving and asserting realpath containment."""
|
|
filename = request.args.get("file", "readme.txt")
|
|
base_dir = "/var/www/files"
|
|
filepath = os.path.join(base_dir, filename)
|
|
real_path = os.path.realpath(filepath)
|
|
if not real_path.startswith(os.path.realpath(base_dir)):
|
|
return "Access denied", 403
|
|
return send_file(real_path)
|
|
|
|
|
|
JWT_SECRET = "super_secret_jwt_key_12345"
|
|
|
|
|
|
def verify_jwt_none_allowed(token: str) -> Dict:
|
|
"""Decode a JWT without verifying its signature."""
|
|
return jwt.decode(token, options={"verify_signature": False})
|
|
|
|
|
|
def verify_jwt_with_secret(token: str, secret: str) -> Dict:
|
|
"""Decode a JWT using the supplied HS256 secret."""
|
|
return jwt.decode(token, secret, algorithms=["HS256"])
|
|
|
|
|
|
@app.route("/fetch/url")
|
|
def fetch_url():
|
|
"""Fetch the bytes at the URL given in the url parameter."""
|
|
url = request.args.get("url")
|
|
response = requests.get(url)
|
|
return response.text
|
|
|
|
|
|
@app.route("/fetch/allowlisted")
|
|
def fetch_allowlisted():
|
|
"""Fetch the URL after asserting its host is on the allowlist."""
|
|
url = request.args.get("url")
|
|
parsed = urlparse(url)
|
|
allowed_hosts = ["api.github.com", "cdn.example.com"]
|
|
if parsed.netloc not in allowed_hosts:
|
|
return "Domain not allowed", 403
|
|
response = requests.get(url)
|
|
return response.text
|
|
|
|
def run_system_command(user_input: str):
|
|
"""Echo the supplied user input via os.system."""
|
|
os.system(f"echo {user_input}")
|
|
|
|
|
|
def run_hardcoded_command():
|
|
"""Run the date binary via os.system."""
|
|
os.system("date")
|
|
|
|
|
|
def generate_token_random() -> str:
|
|
"""Generate a 32-character token using the random module."""
|
|
return "".join(random.choices("abcdefghijklmnopqrstuvwxyz0123456789", k=32))
|
|
|
|
|
|
def shuffle_playlist(items: List[str]) -> List[str]:
|
|
"""Return a shuffled copy of the supplied list."""
|
|
shuffled = items.copy()
|
|
random.shuffle(shuffled)
|
|
return shuffled
|
|
|
|
|
|
DEBUG_MODE = True
|
|
|
|
|
|
@app.route("/debug/eval")
|
|
def debug_eval():
|
|
"""Evaluate the expr query parameter when debug mode is enabled."""
|
|
if DEBUG_MODE:
|
|
expr = request.args.get("expr", "1+1")
|
|
return str(eval(expr))
|
|
return "Disabled"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(debug=True, host="0.0.0.0", port=5001)
|