Gate public endpoint with basic auth via PUBLIC_USER/PUBLIC_PASS env vars
This commit is contained in:
@@ -187,14 +187,40 @@ def retry_dead_letter():
|
|||||||
|
|
||||||
# ── Public file serving ───────────────────────────────────────
|
# ── Public file serving ───────────────────────────────────────
|
||||||
# Serves deployment artifacts for KoreNet/Corné to pull directly
|
# Serves deployment artifacts for KoreNet/Corné to pull directly
|
||||||
# No auth required — public read-only static files
|
# Basic auth gated — credentials in environment variables
|
||||||
|
|
||||||
|
def check_public_auth():
|
||||||
|
"""Basic auth for public file serving."""
|
||||||
|
from flask import request, Response
|
||||||
|
import os, hmac, hashlib
|
||||||
|
auth = request.authorization
|
||||||
|
expected_user = os.environ.get('PUBLIC_USER', 'korenet')
|
||||||
|
expected_pass = os.environ.get('PUBLIC_PASS', '')
|
||||||
|
if not expected_pass:
|
||||||
|
return True # No password configured — open access
|
||||||
|
if not auth:
|
||||||
|
return False
|
||||||
|
# Constant-time comparison
|
||||||
|
user_ok = hmac.compare_digest(auth.username.encode(), expected_user.encode())
|
||||||
|
pass_ok = hmac.compare_digest(auth.password.encode(), expected_pass.encode())
|
||||||
|
return user_ok and pass_ok
|
||||||
|
|
||||||
|
def auth_required():
|
||||||
|
from flask import Response
|
||||||
|
return Response(
|
||||||
|
'Authentication required.',
|
||||||
|
401,
|
||||||
|
{'WWW-Authenticate': 'Basic realm="VZ Public Files"'}
|
||||||
|
)
|
||||||
|
|
||||||
PUBLIC_DIR = os.path.join(os.path.dirname(__file__), 'public')
|
PUBLIC_DIR = os.path.join(os.path.dirname(__file__), 'public')
|
||||||
os.makedirs(PUBLIC_DIR, exist_ok=True)
|
os.makedirs(PUBLIC_DIR, exist_ok=True)
|
||||||
|
|
||||||
@app.route('/public/<path:filename>')
|
@app.route('/public/<path:filename>')
|
||||||
def serve_public(filename):
|
def serve_public(filename):
|
||||||
"""Serve public deployment files."""
|
"""Serve public deployment files — basic auth required."""
|
||||||
|
if not check_public_auth():
|
||||||
|
return auth_required()
|
||||||
import re
|
import re
|
||||||
# Sanitise path — no directory traversal
|
# Sanitise path — no directory traversal
|
||||||
if '..' in filename or filename.startswith('/'):
|
if '..' in filename or filename.startswith('/'):
|
||||||
@@ -218,7 +244,9 @@ def serve_public(filename):
|
|||||||
|
|
||||||
@app.route('/public')
|
@app.route('/public')
|
||||||
def list_public():
|
def list_public():
|
||||||
"""List available public files."""
|
"""List available public files — basic auth required."""
|
||||||
|
if not check_public_auth():
|
||||||
|
return auth_required()
|
||||||
files = []
|
files = []
|
||||||
if os.path.exists(PUBLIC_DIR):
|
if os.path.exists(PUBLIC_DIR):
|
||||||
for f in sorted(os.listdir(PUBLIC_DIR)):
|
for f in sorted(os.listdir(PUBLIC_DIR)):
|
||||||
|
|||||||
Reference in New Issue
Block a user