#!/usr/bin/env python3
"""Fix the corrupted admin.py file."""
import os
import sys
# Read the file
with open("app/routes/admin.py") as f:
lines = f.readlines()
# Find the header (lines 1-34) and the routes (lines 109+)
# Reconstruct the missing helper functions
header_end = None
routes_start = None
for i, line in enumerate(lines):
if line.startswith("def _get_redis"):
header_end = i
break
if line.strip().startswith("def decorated_function"):
routes_start = i
break
print(f"Header ends at line {header_end + 1}")
print(f"Routes start at line {routes_start + 1}")
# Extract header and routes
header = "".join(lines[:header_end])
routes = "".join(lines[routes_start-1:]) # Include the @admin_required line before decorated_function
# Build the helper functions
helpers = '''
def _get_redis():
"""Lazily connect to Redis for persistent brute-force tracking."""
global _redis
if _redis is None:
try:
import redis
_redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)
except Exception:
_redis = False # Sentinel to avoid retrying
return _redis if _redis is not False else None
def _check_admin_rate_limit(ip):
"""Check if IP has exceeded login attempt rate limit.
Returns True if rate limit exceeded, False otherwise.
Gracefully degrades if Redis is unavailable.
"""
r = _get_redis()
if r is None:
return False
try:
key = f"admin_login_attempts:{ip}"
count = r.get(key)
if count and int(count) >= ADMIN_MAX_ATTEMPTS:
return True
return False
except Exception:
return False
def _record_admin_failure(ip):
"""Record a failed login attempt for an IP address.
Tracks per-IP failures with 15-minute expiry.
Gracefully degrades if Redis is unavailable.
"""
r = _get_redis()
if r is None:
return
try:
key = f"admin_login_attempts:{ip}"
r.incr(key)
if r.ttl(key) < 0:
r.expire(key, ADMIN_LOCKOUT_SECONDS)
except Exception:
pass
def _check_session_auth():
"""Check if admin is authenticated via session."""
if not session.get("admin_authenticated"):
return False
if time.time() - session.get("admin_login_time", 0) > 86400:
session.clear()
return False
return True
def check_auth():
"""Unified auth check: session first, then HTTP Basic fallback."""
if _check_session_auth():
return True
if not request.authorization:
return False
try:
return request.authorization.username == ADMIN_USER and request.authorization.password == ADMIN_PASS
except Exception:
return False
def _unauthorized_json():
"""Return 401 JSON response."""
return jsonify({"error": "Authentication required", "status": 401}), 401
def admin_required(f):
"""Decorator that enforces admin authentication."""
from functools import wraps
@wraps(f)
def decorated_function(*args, **kwargs):
if not check_auth():
if request.is_json or request.accept_mimetypes.best == "application/json":
return _unauthorized_json()
return redirect(url_for("admin.login"))
return f(*args, **kwargs)
return decorated_function
'''
# Fix the constants
header = header.replace('_ADMIN_MAX_ATTEMPTS = 10', 'ADMIN_MAX_ATTEMPTS = 10')
header = header.replace('_ADMIN_LOCKOUT_DURATION = 900', 'ADMIN_LOCKOUT_SECONDS = 900 # 15 minutes')
header = header.replace('_ADMIN_WINDOW = 300', '') # Remove unused constant
# Remove old failed attempts dict
header = header.replace('_ADMIN_FAILED_ATTEMPTS = {}', '')
# Reconstruct the file
new_content = header + helpers + routes
with open("app/routes/admin.py", "w") as f:
f.write(new_content)
print("Fixed admin.py successfully")