import functools
import ipaddress
import logging
import os
import time
import uuid
from flask import Blueprint, flash, jsonify, redirect, render_template, request, session, url_for
from app.models import (
TIERS,
add_site,
create_template,
delete_site,
delete_template,
dismiss_spam,
get_admin_stats,
get_monthly_submission_count,
get_site_analytics,
get_site_analytics_summary,
get_spam_submissions,
get_submission,
get_template_by_slug,
get_user_site_count,
list_all_users,
list_sites,
list_submissions,
list_templates,
site_device_stats,
site_geo_stats,
site_impression_stats,
update_template,
)
admin_bp = Blueprint("admin", __name__)
ADMIN_USER = os.environ.get("ADMIN_USER")
if not ADMIN_USER:
raise RuntimeError("ADMIN_USER environment variable is required")
ADMIN_PASS = os.environ.get("ADMIN_PASS")
if ADMIN_PASS is None or ADMIN_PASS == "":
raise RuntimeError("ADMIN_PASS environment variable is required for admin routes")
_admin_logger = logging.getLogger("admin.auth")
# Session-based auth replaces HTTP Basic auth
# ─── Brute-force protection ──────────────────────────────────────────────
_ADMIN_MAX_ATTEMPTS = 10
_ADMIN_LOCKOUT_DURATION = 900 # 15 minutes
_ADMIN_WINDOW = 300 # 5 minute sliding window
_redis = None
_ADMIN_FAILED_ATTEMPTS = {}
def _get_redis():
"""Lazily connect to Redis for persistent brute-force tracking."""
global _redis
if _redis is None:
try:
import redis as r
_redis = r.from_url("redis://localhost:6379/0", decode_responses=True)
_redis.ping() # Test the connection eagerly
except Exception:
_redis = False # Sentinel to avoid retrying
return _redis if _redis is not False else None
def _check_admin_rate_limit(ip):
"""Return True if IP is rate-limited."""
now = time.time()
r = _get_redis()
if r is not None:
key = f"admin:bf:{ip}"
count = r.get(key)
if count and int(count) >= _ADMIN_MAX_ATTEMPTS:
ttl = r.ttl(key)
if ttl and ttl > 0:
return True
return False
# In-memory fallback
if ip in _ADMIN_FAILED_ATTEMPTS:
attempts = [t for t in _ADMIN_FAILED_ATTEMPTS[ip] if now - t < _ADMIN_WINDOW]
_ADMIN_FAILED_ATTEMPTS[ip] = attempts
if len(attempts) >= _ADMIN_MAX_ATTEMPTS:
return True
return False
def _record_admin_failure(ip):
"""Record a failed login attempt for the given IP."""
now = time.time()
r = _get_redis()
if r is not None:
key = f"admin:bf:{ip}"
pipe = r.pipeline()
pipe.incr(key)
pipe.expire(key, _ADMIN_LOCKOUT_DURATION)
pipe.execute()
else:
_ADMIN_FAILED_ATTEMPTS.setdefault(ip, []).append(now)
def _check_session_auth():
"""Check if the current session is authenticated as admin."""
if session.get("admin_authenticated"):
login_time = session.get("admin_login_time", 0)
if time.time() - login_time < 3600: # 1 hour session timeout (Phase 8)
return True
session.clear()
return False
def check_auth():
"""Validate session + password hash."""
return _check_session_auth()
def _unauthorized_json():
"""Return 401 JSON response."""
return jsonify({"error": "Unauthorized"}), 401
def admin_required(f):
@functools.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
# ─── Login routes ────────────────────────────────────────────────────────────
@admin_bp.route("/login", methods=["GET", "POST"])
def login():
if _check_session_auth():
return redirect(url_for("admin.dashboard"))
ip = request.remote_addr
if _check_admin_rate_limit(ip):
return render_template(
"admin_login.html", errors=["Too many failed attempts. Try again in 15 minutes."], email=""
)
if request.method == "POST":
email = request.form.get("email", "").strip()
password = request.form.get("password", "")
if not email or not password:
return render_template("admin_login.html", errors=["Email and password required"], email=email)
password_valid = False
if ADMIN_PASS.startswith("$2b$") or ADMIN_PASS.startswith("$2a$"):
import bcrypt
if bcrypt.checkpw(password.encode("utf-8"), ADMIN_PASS.encode("utf-8")):
password_valid = True
else:
if password == ADMIN_PASS:
password_valid = True
if password_valid:
session.clear()
session["admin_authenticated"] = True
session["admin_login_time"] = time.time()
_admin_logger.info("Admin login successful from %s", ip)
flash("Signed in as admin.", "success")
return redirect(url_for("admin.dashboard"))
_record_admin_failure(ip)
return render_template("admin_login.html", errors=["Invalid credentials"], email=email)
return render_template("admin_login.html", errors=[], email="")
@admin_bp.route("/logout", methods=["POST"])
def logout():
session.clear()
flash("Logged out.", "info")
return redirect(url_for("admin.login"))
# ─── Dashboard ───────────────────────────────────────────────────────────────
@admin_bp.route("/dashboard")
@admin_required
def dashboard():
site_filter = request.args.get("site", type=int)
submissions = list_submissions(limit=50, site_filter=site_filter)
sites = list_sites()
platform_stats = get_admin_stats()
return render_template(
"dashboard.html", submissions=submissions, sites=sites, site_filter=site_filter, platform_stats=platform_stats
)
@admin_bp.route("/sites")
@admin_required
def sites():
return render_template("sites.html", sites=list_sites())
@admin_bp.route("/sites/add", methods=["POST"])
@admin_required
def add_site_route():
name = request.form.get("name", "").strip()
owner_email = request.form.get("owner_email", "").strip()
smtp_from = request.form.get("smtp_from", "").strip()
if not name or not owner_email:
return "Name and owner email are required", 400
site = add_site(name, owner_email, smtp_from or None)
return render_template("site_added.html", site=site)
@admin_bp.route("/sites/delete/<int:site_id>", methods=["POST"])
@admin_required
def delete_site_route(site_id):
delete_site(site_id)
return redirect(url_for("admin.sites"))
@admin_bp.route("/api/sites")
@admin_required
def api_sites():
sites = list_sites()
for s in sites:
token = s.get("token", "")
s["token"] = token[:6] + "****" + token[-4:] if len(token) > 10 else "****"
return jsonify(sites)
@admin_bp.route("/api/submissions")
@admin_required
def api_submissions():
site_filter = request.args.get("site", type=int)
limit = request.args.get("limit", 50, type=int)
return jsonify(list_submissions(limit=limit, site_filter=site_filter))
@admin_bp.route("/users")
@admin_required
def users():
users = list_all_users()
result = []
for u in users:
tier_config = TIERS.get(u["tier"], TIERS["free"])
result.append(
{
**u,
"site_count": get_user_site_count(u["id"]),
"sub_count": get_monthly_submission_count(u["id"]),
"max_sites": tier_config["max_sites"],
"max_subs": tier_config["max_submissions"],
"tier_name": tier_config["name"],
}
)
return render_template("admin_users.html", users=result)
@admin_bp.route("/api/admin/stats")
@admin_required
def admin_stats():
return jsonify(get_admin_stats())
@admin_bp.route("/api/admin/analytics/<int:site_id>")
@admin_required
def admin_site_analytics(site_id):
days = request.args.get("days", 30, type=int)
return jsonify(
{
"daily": get_site_analytics(site_id, days=days),
"summary": get_site_analytics_summary(site_id),
"geo": site_geo_stats(site_id, days=days),
"device": site_device_stats(site_id, days=days),
"impressions": site_impression_stats(site_id, days=days),
}
)
@admin_bp.route("/api/admin/spam")
@admin_required
def admin_spam_list():
limit = request.args.get("limit", 50, type=int)
offset = request.args.get("offset", 0, type=int)
return jsonify(get_spam_submissions(limit=limit, offset=offset))
@admin_bp.route("/api/admin/spam/<int:submission_id>/dismiss", methods=["POST"])
@admin_required
def admin_dismiss_spam(submission_id):
dismiss_spam(submission_id)
return jsonify({"success": True})
@admin_bp.route("/templates")
@admin_required
def admin_templates():
return render_template("admin_templates.html", templates=list_templates())
@admin_bp.route("/api/admin/templates", methods=["GET"])
@admin_required
def admin_templates_api():
return jsonify(list_templates())
@admin_bp.route("/api/admin/templates", methods=["POST"])
@admin_required
def admin_create_template():
data = request.get_json(silent=True)
if not data or not data.get("slug") or not data.get("name"):
return jsonify({"error": "slug and name required"}), 400
import copy
field_config = copy.deepcopy(data.get("field_config", []))
result = create_template(
slug=data["slug"],
name=data["name"],
description=data.get("description", ""),
category=data.get("category", "Business"),
field_config=field_config,
success_message=data.get("success_message"),
is_featured=int(data.get("is_featured", 0)),
created_by=data.get("created_by", "admin"),
)
if not result:
return jsonify({"error": "Template with this slug already exists"}), 409
return jsonify(result), 201
@admin_bp.route("/api/admin/templates/<slug>", methods=["PUT"])
@admin_required
def admin_update_template(slug):
data = request.get_json(silent=True)
if not data:
return jsonify({"error": "JSON body required"}), 400
import copy
if "field_config" in data:
data["field_config"] = copy.deepcopy(data["field_config"])
result = update_template(slug, **data)
if not result:
return jsonify({"error": "Template not found"}), 404
return jsonify(result)
@admin_bp.route("/api/admin/templates/<slug>", methods=["DELETE"])
@admin_required
def admin_delete_template(slug):
if not delete_template(slug):
return jsonify({"error": "Template not found"}), 404
return jsonify({"success": True})
@admin_bp.route("/api/admin/backups", methods=["GET"])
@admin_required
def admin_backups_list():
from app.services.backup import list_backups
return jsonify(list_backups())
@admin_bp.route("/api/admin/backups/now", methods=["POST"])
@admin_required
def admin_backup_now():
from app.services.backup import backup_database
path = backup_database()
if path:
return jsonify({"success": True, "path": path})
return jsonify({"success": False, "error": "Backup failed"}), 500
@admin_bp.route("/api/admin/backups/restore", methods=["POST"])
@admin_required
def admin_restore_backup():
data = request.get_json(silent=True)
if not data or not data.get("backup_path"):
return jsonify({"error": "backup_path required in JSON body"}), 400
from app.services.backup import list_backups, restore_backup
backups = list_backups()
known_paths = {b["path"] for b in backups}
if data["backup_path"] not in known_paths:
return jsonify({"error": "Backup not found"}), 404
success = restore_backup(data["backup_path"])
if success:
return jsonify({"success": True, "restored_from": data["backup_path"]})
return jsonify({"success": False, "error": "Restore failed"}), 500