#!/usr/bin/env python3
"""
BNKFX Admin Module for Preview Server
Handles Payment Methods, User Management, Impersonation, KYC & Balance Adjustments.
"""
import json
import time
import urllib.parse
import hashlib
import html as html_lib
import mail_service
import platform_settings_service
def handle_admin_post(handler, parsed, data, STATE):
path = parsed.path
# 1. Create Payment Method
if path == "/admin/payment-methods":
new_id = len(STATE.get("payment_methods", [])) + 1
code = str(data.get("code", "USDC")).upper().strip()
network = str(data.get("network", "ERC20")).upper().strip()
name = str(data.get("name", f"{code} ({network})")).strip()
wallet = str(data.get("wallet_address", "")).strip()
# Determine icon URL (use uploaded base64 data, custom icon_url or fallback to SVG)
icon_data = data.get("icon_data", "")
icon_mime = data.get("icon_mime", "image/png")
if icon_data:
icon_url = f"/admin/uploads/icon/{new_id}"
else:
icon_url = data.get("icon_url") or f"/assets/images/crypto/{code.lower()}.svg"
# QR code settings
qr_enabled = str(data.get("qr_enabled", "1")) == "1"
qr_type = str(data.get("qr_type", "auto")) # "auto", "upload", "none"
qr_data = data.get("qr_data", "") # base64 uploaded QR if qr_type == "upload"
try:
min_dep = float(data.get("min_deposit", 50.0))
except (ValueError, TypeError):
min_dep = 50.0
try:
max_dep = float(data.get("max_deposit", 100000.0))
except (ValueError, TypeError):
max_dep = 100000.0
try:
confs = int(data.get("required_confirmations", 12))
except (ValueError, TypeError):
confs = 12
try:
fee_pct = float(data.get("processing_fee_pct", 0.0))
except (ValueError, TypeError):
fee_pct = 0.0
try:
sort_order = int(data.get("sort_order", new_id))
except (ValueError, TypeError):
sort_order = new_id
new_pm = {
"id": new_id,
"name": name,
"code": code,
"network": network,
"icon_url": icon_url,
"icon_data": icon_data if icon_data else "",
"icon_mime": icon_mime,
"wallet_address": wallet,
"min_deposit": min_dep,
"max_deposit": max_dep,
"required_confirmations": confs,
"processing_fee_pct": fee_pct,
"instructions": str(data.get("instructions", "")).strip(),
"status": str(data.get("status", "active")),
"sort_order": sort_order,
"payment_type": str(data.get("payment_type", "cryptocurrency")),
"processing_time": str(data.get("processing_time", "3 - 8 mins")).strip() or "3 - 8 mins",
"notes": str(data.get("notes", "")).strip(),
"qr_enabled": qr_enabled,
"qr_type": qr_type,
"qr_data": qr_data
}
STATE.setdefault("payment_methods", []).append(new_pm)
STATE.setdefault("audit_logs", []).append({
"action": "admin_payment_method_created",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Created payment method {new_pm['name']} ({code} {network})"
})
handler.send_response(302)
handler.send_header("Location", "/admin/payment-methods")
handler.end_headers()
return True
# 2. Update Payment Method
if path.startswith("/admin/payment-methods/") and path.endswith("/update"):
pm_id = int(path.split("/")[3])
pm = next((p for p in STATE.get("payment_methods", []) if p["id"] == pm_id), None)
if pm:
new_addr = str(data.get("wallet_address", pm["wallet_address"])).strip()
if new_addr and new_addr != pm["wallet_address"]:
STATE.setdefault("wallet_address_history", []).append({
"payment_method_id": pm_id,
"old_address": pm["wallet_address"],
"new_address": new_addr,
"reason": data.get("change_reason", "Administrative wallet rotation"),
"changed_by": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S")
})
STATE.setdefault("audit_logs", []).append({
"action": "admin_wallet_address_updated",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Address updated for {pm['name']}: {new_addr}"
})
pm["wallet_address"] = new_addr
if data.get("name"):
pm["name"] = str(data["name"]).strip()
if data.get("code"):
pm["code"] = str(data["code"]).upper().strip()
if data.get("network"):
pm["network"] = str(data["network"]).upper().strip()
if "min_deposit" in data and str(data["min_deposit"]).strip() != "":
try:
pm["min_deposit"] = float(data["min_deposit"])
except (ValueError, TypeError):
pass
if "max_deposit" in data and str(data["max_deposit"]).strip() != "":
try:
pm["max_deposit"] = float(data["max_deposit"])
except (ValueError, TypeError):
pass
if "required_confirmations" in data and str(data["required_confirmations"]).strip() != "":
try:
pm["required_confirmations"] = int(data["required_confirmations"])
except (ValueError, TypeError):
pass
if "processing_fee_pct" in data and str(data["processing_fee_pct"]).strip() != "":
try:
pm["processing_fee_pct"] = float(data["processing_fee_pct"])
except (ValueError, TypeError):
pass
if "instructions" in data:
pm["instructions"] = str(data.get("instructions", "")).strip()
if "status" in data and data["status"]:
pm["status"] = str(data["status"])
if "processing_time" in data and str(data["processing_time"]).strip():
pm["processing_time"] = str(data["processing_time"]).strip()
if "sort_order" in data and str(data["sort_order"]).strip() != "":
try:
pm["sort_order"] = int(data["sort_order"])
except (ValueError, TypeError):
pass
if "notes" in data:
pm["notes"] = str(data.get("notes", "")).strip()
# Preserve existing icon if no new icon uploaded
if data.get("icon_data"):
pm["icon_data"] = data["icon_data"]
pm["icon_mime"] = data.get("icon_mime", "image/png")
pm["icon_url"] = f"/admin/uploads/icon/{pm_id}"
elif data.get("icon_url"):
pm["icon_url"] = data["icon_url"]
# QR Code preservation & update
if "qr_enabled" in data:
pm["qr_enabled"] = str(data.get("qr_enabled", "1")) == "1"
if "qr_type" in data:
pm["qr_type"] = str(data.get("qr_type", pm.get("qr_type", "auto")))
if data.get("qr_data"):
pm["qr_data"] = data["qr_data"]
STATE.setdefault("audit_logs", []).append({
"action": "admin_payment_method_updated",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Payment method #{pm_id} ({pm['name']}) updated: limits ${pm['min_deposit']:.2f}-${pm['max_deposit']:.2f}, status {pm['status']}"
})
handler.send_response(302)
handler.send_header("Location", "/admin/payment-methods")
handler.end_headers()
return True
# 3. Toggle Status Payment Method
if path.startswith("/admin/payment-methods/") and path.endswith("/toggle-status"):
pm_id = int(path.split("/")[3])
pm = next((p for p in STATE.get("payment_methods", []) if p["id"] == pm_id), None)
if pm:
pm["status"] = "disabled" if pm["status"] == "active" else "active"
STATE.setdefault("audit_logs", []).append({
"action": "admin_payment_method_status_toggled",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Method {pm['name']} toggled to {pm['status']}"
})
handler.send_response(302)
handler.send_header("Location", "/admin/payment-methods")
handler.end_headers()
return True
# 4. Delete / Archive Payment Method
if path.startswith("/admin/payment-methods/") and path.endswith("/delete"):
pm_id = int(path.split("/")[3])
pm = next((p for p in STATE.get("payment_methods", []) if p["id"] == pm_id), None)
if pm:
pm["status"] = "archived"
STATE.setdefault("audit_logs", []).append({
"action": "admin_payment_method_archived",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Payment method {pm['name']} archived"
})
handler.send_response(302)
handler.send_header("Location", "/admin/payment-methods")
handler.end_headers()
return True
# 5. User Update Profile
if path.startswith("/admin/users/") and path.endswith("/update-profile"):
uid = int(path.split("/")[3])
target = next((u for u in STATE.get("users", []) if u["id"] == uid), None)
if target:
target["first_name"] = str(data.get("first_name", target["first_name"]))
target["last_name"] = str(data.get("last_name", target["last_name"]))
target["phone"] = str(data.get("phone", target.get("phone", "")))
target["country"] = str(data.get("country", target.get("country", "")))
target["address"] = str(data.get("address", target.get("address", "")))
if STATE.get("user", {}).get("id") == uid:
STATE["user"].update(target)
STATE.setdefault("audit_logs", []).append({
"action": "admin_user_profile_updated",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Updated profile for user #{uid} ({target['email']})"
})
handler.send_response(302)
handler.send_header("Location", f"/admin/users/{uid}?tab=profile")
handler.end_headers()
return True
# 6. User Suspend / Reactivate
if path.startswith("/admin/users/") and path.endswith("/toggle-status"):
uid = int(path.split("/")[3])
target = next((u for u in STATE.get("users", []) if u["id"] == uid), None)
if target:
if target["status"] == "active":
target["status"] = "suspended"
target["suspended_reason"] = str(data.get("suspend_reason", "Suspended by administrator"))
target["suspended_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
else:
target["status"] = "active"
target["suspended_reason"] = None
target["suspended_at"] = None
if STATE.get("user", {}).get("id") == uid:
STATE["user"]["status"] = target["status"]
STATE["user"]["suspended_reason"] = target["suspended_reason"]
STATE["user"]["suspended_at"] = target["suspended_at"]
STATE.setdefault("audit_logs", []).append({
"action": f"admin_user_{target['status']}",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"User #{uid} status changed to {target['status']}"
})
handler.send_response(302)
handler.send_header("Location", f"/admin/users/{uid}?tab=status")
handler.end_headers()
return True
# 7. User Verify Email
if path.startswith("/admin/users/") and path.endswith("/verify-email"):
uid = int(path.split("/")[3])
target = next((u for u in STATE.get("users", []) if u["id"] == uid), None)
if target:
target["email_verified_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
if STATE.get("user", {}).get("id") == uid:
STATE["user"]["email_verified_at"] = target["email_verified_at"]
STATE.setdefault("audit_logs", []).append({
"action": "admin_email_verified",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Email verified manually for user #{uid} ({target['email']})"
})
handler.send_response(302)
handler.send_header("Location", f"/admin/users/{uid}?tab=email")
handler.end_headers()
return True
# 8. User Revoke Email
if path.startswith("/admin/users/") and path.endswith("/revoke-email"):
uid = int(path.split("/")[3])
target = next((u for u in STATE.get("users", []) if u["id"] == uid), None)
if target:
target["email_verified_at"] = None
if STATE.get("user", {}).get("id") == uid:
STATE["user"]["email_verified_at"] = None
STATE.setdefault("audit_logs", []).append({
"action": "admin_email_verification_revoked",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Email verification revoked for user #{uid} ({target['email']})"
})
handler.send_response(302)
handler.send_header("Location", f"/admin/users/{uid}?tab=email")
handler.end_headers()
return True
# 9. User KYC Review
if path.startswith("/admin/users/") and path.endswith("/kyc"):
uid = int(path.split("/")[3])
action = str(data.get("action", "approve"))
reason = str(data.get("reason", "KYC criteria not met"))
target = next((u for u in STATE.get("users", []) if u["id"] == uid), None)
if target:
if action == "approve":
target["kyc_status"] = "approved"
target["kyc_reviewed_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
target["kyc_rejection_reason"] = None
else:
target["kyc_status"] = "rejected"
target["kyc_reviewed_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
target["kyc_rejection_reason"] = reason
if STATE.get("user", {}).get("id") == uid:
STATE["user"]["kyc_status"] = target["kyc_status"]
STATE["user"]["kyc_rejection_reason"] = target["kyc_rejection_reason"]
STATE.setdefault("audit_logs", []).append({
"action": f"admin_kyc_{action}",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"KYC {action} for user #{uid} ({target['email']})"
})
handler.send_response(302)
handler.send_header("Location", f"/admin/users/{uid}?tab=kyc")
handler.end_headers()
return True
# 10. Balance Adjustment (Double-Entry Ledger)
if path.startswith("/admin/users/") and path.endswith("/adjust-balance"):
uid = int(path.split("/")[3])
direction = str(data.get("direction", "credit")).lower()
amt = float(data.get("amount", 0.0))
reason = str(data.get("reason", "Administrative balance adjustment"))
allow_overdraft = bool(data.get("allow_overdraft", False))
target = next((u for u in STATE.get("users", []) if u["id"] == uid), None)
if direction == "debit" and STATE["accounts"][0]["balance"] < amt and not allow_overdraft:
res = {"success": False, "error": f"Insufficient funds for debit. Current balance is ${STATE['accounts'][0]['balance']:.2f} USD."}
handler.send_response(400)
handler.send_header("Content-Type", "application/json")
handler.end_headers()
handler.wfile.write(json.dumps(res).encode('utf-8'))
return True
if direction == "credit":
STATE["accounts"][0]["balance"] += amt
else:
STATE["accounts"][0]["balance"] -= amt
ref = "ADJ-" + hashlib.sha256(str(time.time()).encode()).hexdigest()[:8].upper()
adj_tx = {
"ref": ref,
"direction": direction,
"sender": "System Operational Reserve Vault",
"sender_email": "vault@bnkfx.internal",
"recipient": f"{target['first_name']} {target['last_name']}" if target else "Customer",
"recipient_email": target["email"] if target else "customer@bnkfx.com",
"amount": amt,
"currency": "USD",
"fee": 0.00,
"status": "Completed",
"date": "Just now",
"reason": reason
}
STATE["transfers"].insert(0, adj_tx)
STATE.setdefault("audit_logs", []).append({
"action": f"admin_balance_{direction}",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Balance {direction} of ${amt:.2f} USD posted for user #{uid}. New balance: ${STATE['accounts'][0]['balance']:.2f}"
})
handler.send_response(302)
handler.send_header("Location", f"/admin/users/{uid}?tab=balance")
handler.end_headers()
return True
# 11. Impersonate User
if path.startswith("/admin/impersonate/") and not path.endswith("/exit"):
uid = int(path.split("/")[3])
target = next((u for u in STATE.get("users", []) if u["id"] == uid), None)
if target:
STATE["impersonation"] = {
"original_admin_id": 1,
"original_admin_email": "admin@bnkfx.com",
"impersonated_user_id": target["id"],
"impersonated_user_email": target["email"],
"started_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"reason": str(data.get("impersonate_reason", "Customer troubleshooting"))
}
STATE["user"] = target
STATE.setdefault("audit_logs", []).append({
"action": "admin_user_impersonation_started",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Support session started: logged in as {target['email']}"
})
handler.send_response(302)
handler.send_header("Location", "/dashboard")
handler.end_headers()
return True
# 10. KYC Management: Start Review
if path.startswith("/admin/kyc/") and path.endswith("/start-review"):
sub_id = int(path.split("/")[3])
sub = next((k for k in STATE.get("kyc_verifications", []) if k["id"] == sub_id), None)
if sub:
sub["status"] = "under_review"
sub["updated_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
target_user = next((u for u in STATE.get("users", []) if u["id"] == sub["user_id"]), None)
if target_user:
target_user["kyc_status"] = "under_review"
if STATE.get("user", {}).get("id") == target_user["id"]:
STATE["user"]["kyc_status"] = "under_review"
STATE.setdefault("audit_logs", []).append({
"action": "admin_kyc_start_review",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Compliance review started for KYC submission #{sub_id} (User #{sub['user_id']})"
})
handler.send_response(302)
handler.send_header("Location", "/admin/kyc")
handler.end_headers()
return True
# 11. KYC Management: Approve
if path.startswith("/admin/kyc/") and path.endswith("/approve"):
sub_id = int(path.split("/")[3])
sub = next((k for k in STATE.get("kyc_verifications", []) if k["id"] == sub_id), None)
if sub:
sub["status"] = "approved"
sub["reviewed_by"] = 1
sub["reviewed_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
sub["updated_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
sub["rejection_reason"] = None
notes = str(data.get("notes", "")).strip()
if notes:
sub["notes"] = notes
target_user = next((u for u in STATE.get("users", []) if u["id"] == sub["user_id"]), None)
if target_user:
target_user["kyc_status"] = "approved"
target_user["kyc_reviewed_at"] = sub["reviewed_at"]
target_user["kyc_rejection_reason"] = None
if STATE.get("user", {}).get("id") == target_user["id"]:
STATE["user"]["kyc_status"] = "approved"
STATE["user"]["kyc_reviewed_at"] = sub["reviewed_at"]
STATE["user"]["kyc_rejection_reason"] = None
for acc in STATE.get("accounts", []):
acc["verification_status"] = "verified"
# Record approval email
user_email = target_user["email"] if target_user else "customer@bnkfx.com"
user_name = f"{target_user['first_name']} {target_user['last_name']}" if target_user else "Customer"
STATE.setdefault("email_logs", []).append({
"id": len(STATE.get("email_logs", [])) + 1,
"user_id": sub["user_id"],
"recipient": user_email,
"subject": "Your Identity Verification Is Complete",
"template": "kyc_approved",
"status": "sent",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S")
})
STATE.setdefault("audit_logs", []).append({
"action": "admin_kyc_approved",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Approved KYC submission #{sub_id} for {user_name} ({user_email})"
})
handler.send_response(302)
handler.send_header("Location", "/admin/kyc")
handler.end_headers()
return True
# 12. KYC Management: Reject
if path.startswith("/admin/kyc/") and path.endswith("/reject"):
sub_id = int(path.split("/")[3])
sub = next((k for k in STATE.get("kyc_verifications", []) if k["id"] == sub_id), None)
reason = str(data.get("reason", "Incomplete or unverified identity documentation")).strip()
if sub:
sub["status"] = "rejected"
sub["rejection_reason"] = reason
sub["reviewed_by"] = 1
sub["reviewed_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
sub["updated_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
target_user = next((u for u in STATE.get("users", []) if u["id"] == sub["user_id"]), None)
if target_user:
target_user["kyc_status"] = "rejected"
target_user["kyc_rejection_reason"] = reason
target_user["kyc_reviewed_at"] = sub["reviewed_at"]
if STATE.get("user", {}).get("id") == target_user["id"]:
STATE["user"]["kyc_status"] = "rejected"
STATE["user"]["kyc_rejection_reason"] = reason
STATE["user"]["kyc_reviewed_at"] = sub["reviewed_at"]
# Record rejection email
user_email = target_user["email"] if target_user else "customer@bnkfx.com"
user_name = f"{target_user['first_name']} {target_user['last_name']}" if target_user else "Customer"
STATE.setdefault("email_logs", []).append({
"id": len(STATE.get("email_logs", [])) + 1,
"user_id": sub["user_id"],
"recipient": user_email,
"subject": "Identity Verification Requires Attention — Alliance Bank",
"template": "kyc_rejected",
"status": "sent",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S")
})
STATE.setdefault("audit_logs", []).append({
"action": "admin_kyc_rejected",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Rejected KYC submission #{sub_id} for {user_name} (Reason: {reason})"
})
handler.send_response(302)
handler.send_header("Location", "/admin/kyc")
handler.end_headers()
return True
# 13. KYC Management: Request Additional Info
if path.startswith("/admin/kyc/") and path.endswith("/request-info"):
sub_id = int(path.split("/")[3])
sub = next((k for k in STATE.get("kyc_verifications", []) if k["id"] == sub_id), None)
notes = str(data.get("notes", "Please upload a clearer image of your document.")).strip()
if sub:
sub["status"] = "additional_info_required"
sub["notes"] = notes
sub["reviewed_by"] = 1
sub["reviewed_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
sub["updated_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
target_user = next((u for u in STATE.get("users", []) if u["id"] == sub["user_id"]), None)
if target_user:
target_user["kyc_status"] = "additional_info_required"
target_user["kyc_notes"] = notes
target_user["kyc_reviewed_at"] = sub["reviewed_at"]
if STATE.get("user", {}).get("id") == target_user["id"]:
STATE["user"]["kyc_status"] = "additional_info_required"
STATE["user"]["kyc_notes"] = notes
STATE["user"]["kyc_reviewed_at"] = sub["reviewed_at"]
# Record email
user_email = target_user["email"] if target_user else "customer@bnkfx.com"
user_name = f"{target_user['first_name']} {target_user['last_name']}" if target_user else "Customer"
STATE.setdefault("email_logs", []).append({
"id": len(STATE.get("email_logs", [])) + 1,
"user_id": sub["user_id"],
"recipient": user_email,
"subject": "Additional Information Required for Verification",
"template": "kyc_additional_info",
"status": "sent",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S")
})
STATE.setdefault("audit_logs", []).append({
"action": "admin_kyc_request_info",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Requested additional info on KYC submission #{sub_id} for {user_name} ({notes})"
})
handler.send_response(302)
handler.send_header("Location", "/admin/kyc")
handler.end_headers()
return True
# 5. Create Stock
if path == "/admin/stocks":
new_id = len(STATE.get("stocks", [])) + 1
name = str(data.get("name", "")).strip()
ticker = str(data.get("ticker", "")).strip().upper()
category = str(data.get("category", "Technology")).strip()
desc = str(data.get("short_description", "")).strip()
def_plan = int(data.get("default_plan_id") or 3)
status = str(data.get("status", "active")).strip()
order = int(data.get("display_order") or 0)
logo_url = str(data.get("logo_url") or f"/assets/images/stocks/{ticker.lower()}.svg").strip()
notes = str(data.get("notes", "")).strip()
new_stk = {
"id": new_id,
"name": name,
"ticker": ticker,
"category": category,
"short_description": desc,
"default_plan_id": def_plan,
"status": status,
"logo_url": logo_url,
"display_order": order,
"notes": notes
}
STATE.setdefault("stocks", []).append(new_stk)
STATE.setdefault("audit_logs", []).append({
"action": "admin_stock_created",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Created stock {ticker} ({name})"
})
handler.send_response(302)
handler.send_header("Location", "/admin/stocks")
handler.end_headers()
return True
# 6. Update Stock
if path.startswith("/admin/stocks/") and path.endswith("/update"):
parts = path.strip("/").split("/")
stk_id = int(parts[2])
stk = next((s for s in STATE.get("stocks", []) if s["id"] == stk_id), None)
if stk:
if "name" in data: stk["name"] = str(data["name"]).strip()
if "ticker" in data: stk["ticker"] = str(data["ticker"]).strip().upper()
if "category" in data: stk["category"] = str(data["category"]).strip()
if "short_description" in data: stk["short_description"] = str(data["short_description"]).strip()
if "default_plan_id" in data and data["default_plan_id"]: stk["default_plan_id"] = int(data["default_plan_id"])
if "status" in data: stk["status"] = str(data["status"]).strip()
if "display_order" in data: stk["display_order"] = int(data["display_order"] or 0)
if "logo_url" in data and data["logo_url"]: stk["logo_url"] = str(data["logo_url"]).strip()
if "notes" in data: stk["notes"] = str(data["notes"]).strip()
STATE.setdefault("audit_logs", []).append({
"action": "admin_stock_updated",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Updated stock {stk['ticker']} (ID: {stk_id})"
})
handler.send_response(302)
handler.send_header("Location", "/admin/stocks")
handler.end_headers()
return True
# 7. Toggle Stock Status
if path.startswith("/admin/stocks/") and path.endswith("/toggle-status"):
parts = path.strip("/").split("/")
stk_id = int(parts[2])
stk = next((s for s in STATE.get("stocks", []) if s["id"] == stk_id), None)
if stk:
stk["status"] = "inactive" if stk["status"] == "active" else "active"
STATE.setdefault("audit_logs", []).append({
"action": "admin_stock_status_toggled",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Toggled status of stock {stk['ticker']} to {stk['status']}"
})
handler.send_response(302)
handler.send_header("Location", "/admin/stocks")
handler.end_headers()
return True
# 8. Delete Stock
if path.startswith("/admin/stocks/") and path.endswith("/delete"):
parts = path.strip("/").split("/")
stk_id = int(parts[2])
stk = next((s for s in STATE.get("stocks", []) if s["id"] == stk_id), None)
if stk:
STATE["stocks"] = [s for s in STATE.get("stocks", []) if s["id"] != stk_id]
STATE.setdefault("audit_logs", []).append({
"action": "admin_stock_deleted",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Deleted stock {stk['ticker']} (ID: {stk_id})"
})
handler.send_response(302)
handler.send_header("Location", "/admin/stocks")
handler.end_headers()
return True
# 9. Create Plan
if path == "/admin/plans":
new_id = len(STATE.get("investment_plans", [])) + 1
name = str(data.get("name", "")).strip()
slug = name.lower().replace(" ", "-")
min_amt = float(data.get("min_amount") or 50.0)
is_unl = str(data.get("is_unlimited", "0")) in ["1", "true", "True"] or (data.get("max_amount") in ["", None])
max_amt = None if is_unl else float(data.get("max_amount") or 100.0)
duration = int(data.get("duration_hours") or 48)
ret_pct = float(data.get("return_percentage") or 20.0)
is_rec = str(data.get("is_recommended", "0")) in ["1", "true", "True"]
new_plan = {
"id": new_id,
"name": name,
"slug": slug,
"description": f"{name} yield product tier.",
"min_amount": min_amt,
"max_amount": max_amt,
"is_unlimited": is_unl,
"duration_hours": duration,
"return_percentage": ret_pct,
"is_recommended": is_rec,
"status": "active",
"display_order": new_id
}
STATE.setdefault("investment_plans", []).append(new_plan)
STATE.setdefault("audit_logs", []).append({
"action": "admin_plan_created",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Created plan {name} (+{ret_pct}%, {duration}h)"
})
handler.send_response(302)
handler.send_header("Location", "/admin/plans")
handler.end_headers()
return True
# 10. Update Plan
if path.startswith("/admin/plans/") and path.endswith("/update"):
parts = path.strip("/").split("/")
plan_id = int(parts[2])
plan = next((p for p in STATE.get("investment_plans", []) if p["id"] == plan_id), None)
if plan:
if "name" in data: plan["name"] = str(data["name"]).strip()
if "min_amount" in data and data["min_amount"]: plan["min_amount"] = float(data["min_amount"])
is_unl = str(data.get("is_unlimited", "0")) in ["1", "true", "True"] or (data.get("max_amount") in ["", None])
plan["is_unlimited"] = is_unl
plan["max_amount"] = None if is_unl else (float(data["max_amount"]) if data.get("max_amount") else None)
if "duration_hours" in data and data["duration_hours"]: plan["duration_hours"] = int(data["duration_hours"])
if "return_percentage" in data and data["return_percentage"]: plan["return_percentage"] = float(data["return_percentage"])
if "is_recommended" in data: plan["is_recommended"] = str(data["is_recommended"]) in ["1", "true", "True"]
STATE.setdefault("audit_logs", []).append({
"action": "admin_plan_updated",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Updated plan {plan['name']} (ID: {plan_id})"
})
handler.send_response(302)
handler.send_header("Location", "/admin/plans")
handler.end_headers()
return True
# 11. Toggle Plan Status
if path.startswith("/admin/plans/") and path.endswith("/toggle-status"):
parts = path.strip("/").split("/")
plan_id = int(parts[2])
plan = next((p for p in STATE.get("investment_plans", []) if p["id"] == plan_id), None)
if plan:
plan["status"] = "inactive" if plan["status"] == "active" else "active"
STATE.setdefault("audit_logs", []).append({
"action": "admin_plan_status_toggled",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Toggled status of plan {plan['name']} to {plan['status']}"
})
handler.send_response(302)
handler.send_header("Location", "/admin/plans")
handler.end_headers()
return True
# 12. Create Card Product
if path == "/admin/cards/products":
new_id = max([p["id"] for p in STATE.get("card_products", [])] + [0]) + 1
name = str(data.get("name", "New Card Product")).strip()
slug = str(data.get("slug", name.lower().replace(" ", "-"))).strip()
card_type = str(data.get("card_type", "virtual")).strip()
network = str(data.get("network", "Visa")).strip()
design = str(data.get("card_design", "obsidian")).strip()
fee = float(data.get("issuance_fee", 0.0))
min_bal = float(data.get("min_balance_required", 0.0))
daily_lim = float(data.get("daily_limit", 5000.0))
monthly_lim = float(data.get("monthly_limit", 25000.0))
description = str(data.get("description", "")).strip()
is_active = int(data.get("is_active", 1))
feats_raw = data.get("features", "")
if isinstance(feats_raw, list):
features = feats_raw
else:
features = [f.strip() for f in str(feats_raw).replace("\r", "").split("\n") if f.strip()]
new_prod = {
"id": new_id,
"name": name,
"slug": slug,
"type": card_type,
"card_type": card_type,
"network": network,
"design": design,
"card_design": design,
"fee": fee,
"issuance_fee": fee,
"min_balance_required": min_bal,
"limit": daily_lim,
"daily_limit": daily_lim,
"monthly_limit": monthly_lim,
"description": description,
"features": features,
"is_active": is_active
}
STATE.setdefault("card_products", []).append(new_prod)
STATE.setdefault("audit_logs", []).append({
"action": "admin_card_product_created",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Created card product '{name}' (ID: {new_id}, Fee: ${fee:.2f})"
})
handler.send_response(302)
handler.send_header("Location", "/admin/cards?tab=products")
handler.end_headers()
return True
# 13. Update Card Product
if path.startswith("/admin/cards/products/") and path.endswith("/update"):
parts = path.strip("/").split("/")
prod_id = int(parts[3])
prod = next((p for p in STATE.get("card_products", []) if p["id"] == prod_id), None)
if prod:
if "name" in data: prod["name"] = str(data["name"]).strip()
if "card_type" in data: prod["card_type"] = prod["type"] = str(data["card_type"]).strip()
if "network" in data: prod["network"] = str(data["network"]).strip()
if "card_design" in data: prod["card_design"] = prod["design"] = str(data["card_design"]).strip()
if "issuance_fee" in data: prod["issuance_fee"] = prod["fee"] = float(data["issuance_fee"])
if "min_balance_required" in data: prod["min_balance_required"] = float(data["min_balance_required"])
if "daily_limit" in data: prod["daily_limit"] = prod["limit"] = float(data["daily_limit"])
if "monthly_limit" in data: prod["monthly_limit"] = float(data["monthly_limit"])
if "description" in data: prod["description"] = str(data["description"]).strip()
if "is_active" in data: prod["is_active"] = int(data["is_active"])
if "features" in data:
feats_raw = data["features"]
if isinstance(feats_raw, list):
prod["features"] = feats_raw
else:
prod["features"] = [f.strip() for f in str(feats_raw).replace("\r", "").split("\n") if f.strip()]
STATE.setdefault("audit_logs", []).append({
"action": "admin_card_product_updated",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Updated card product '{prod['name']}' (ID: {prod_id})"
})
handler.send_response(302)
handler.send_header("Location", "/admin/cards?tab=products")
handler.end_headers()
return True
# 14. Toggle Card Product Status
if path.startswith("/admin/cards/products/") and path.endswith("/toggle-status"):
parts = path.strip("/").split("/")
prod_id = int(parts[3])
prod = next((p for p in STATE.get("card_products", []) if p["id"] == prod_id), None)
if prod:
prod["is_active"] = 0 if prod.get("is_active", 1) == 1 else 1
status_str = "activated" if prod["is_active"] == 1 else "deactivated"
STATE.setdefault("audit_logs", []).append({
"action": f"admin_card_product_{status_str}",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"{status_str.capitalize()} card product '{prod['name']}' (ID: {prod_id})"
})
handler.send_response(302)
handler.send_header("Location", "/admin/cards?tab=products")
handler.end_headers()
return True
# 15. Approve Card Application
if (path.startswith("/admin/cards/applications/") or path.startswith("/admin/cards/orders/")) and path.endswith("/approve"):
parts = path.strip("/").split("/")
order_id = int(parts[3])
order = next((o for o in STATE.get("card_orders", []) if o["id"] == order_id), None)
if order and order.get("status") in ["pending", "processing"]:
order["status"] = "completed"
order["approved_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
# Issue the actual card into STATE["cards"]
new_id = max([c["id"] for c in STATE.get("cards", [])] + [0]) + 1
import random
last_4 = str(random.randint(1000, 9999))
holder_name = str(order.get("user_name", "ELOCHUKWU ELOM")).upper()
prod = next((p for p in STATE.get("card_products", []) if p["id"] == order.get("product_id")), None)
daily_lim = prod["limit"] if prod else 5000.0
issued = {
"id": new_id,
"name": order.get("product_name", "Bnkfx Card"),
"product_name": order.get("product_name", "Bnkfx Card"),
"type": order.get("card_type", "virtual"),
"card_type": order.get("card_type", "virtual"),
"network": order.get("network", "Visa"),
"design": order.get("card_design", "obsidian"),
"card_design": order.get("card_design", "obsidian"),
"masked_number": f"•••• •••• •••• {last_4}",
"card_number_masked": f"•••• •••• •••• {last_4}",
"last_four": last_4,
"exp": f"{random.randint(1, 12):02d}/31",
"expiry_month": 10,
"expiry_year": 2031,
"holder": holder_name,
"card_holder_name": holder_name,
"status": "active",
"daily_limit": daily_lim,
"spent_today": 0.0,
"currency": "USD",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S")
}
STATE.setdefault("cards", []).append(issued)
STATE.setdefault("audit_logs", []).append({
"action": "admin_card_application_approved",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Approved card order {order.get('reference', order_id)} and issued {issued['name']} ending in {last_4}"
})
if handler.headers.get("X-Requested-With") == "XMLHttpRequest" or "application/json" in handler.headers.get("Accept", ""):
handler.send_response(200)
handler.send_header("Content-Type", "application/json")
handler.end_headers()
handler.wfile.write(json.dumps({"success": True, "message": "Card application approved and issued successfully."}).encode('utf-8'))
return True
handler.send_response(302)
handler.send_header("Location", "/admin/cards?tab=applications")
handler.end_headers()
return True
# 16. Reject Card Application
if (path.startswith("/admin/cards/applications/") or path.startswith("/admin/cards/orders/")) and path.endswith("/reject"):
parts = path.strip("/").split("/")
order_id = int(parts[3])
order = next((o for o in STATE.get("card_orders", []) if o["id"] == order_id), None)
if order and order.get("status") in ["pending", "processing"]:
reason = str(data.get("reason", "Application did not meet compliance criteria.")).strip()
order["status"] = "rejected"
order["rejection_reason"] = reason
order["rejected_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
# Refund fee if fee was paid
fee_paid = float(order.get("payment_amount", 0.0))
if fee_paid > 0:
STATE["accounts"][0]["balance"] += fee_paid
STATE.setdefault("transactions", []).insert(0, {
"id": len(STATE.get("transactions", [])) + 1,
"type": "credit",
"category": "card_refund",
"amount": fee_paid,
"currency": "USD",
"status": "completed",
"reference": f"REF-CARD-{order_id}",
"description": f"Refund: Card Issuance Fee - {order.get('product_name', 'Virtual Card')}",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S")
})
STATE.setdefault("audit_logs", []).append({
"action": "admin_card_application_rejected",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Rejected card order {order.get('reference', order_id)}. Reason: {reason}. Refund: ${fee_paid:.2f}"
})
if handler.headers.get("X-Requested-With") == "XMLHttpRequest" or "application/json" in handler.headers.get("Accept", ""):
handler.send_response(200)
handler.send_header("Content-Type", "application/json")
handler.end_headers()
handler.wfile.write(json.dumps({"success": True, "message": "Card application rejected and fee refunded."}).encode('utf-8'))
return True
handler.send_response(302)
handler.send_header("Location", "/admin/cards?tab=applications")
handler.end_headers()
return True
# 13. Update Email Settings
if path in ["/admin/settings/email", "/admin/email-settings"]:
settings = mail_service.get_email_settings(STATE)
# Text fields
if "smtp_host" in data and str(data["smtp_host"]).strip():
settings["smtp_host"] = str(data["smtp_host"]).strip()
if "smtp_port" in data:
try:
settings["smtp_port"] = int(data["smtp_port"])
except (ValueError, TypeError):
pass
if "smtp_username" in data and str(data["smtp_username"]).strip():
settings["smtp_username"] = str(data["smtp_username"]).strip()
# Password handling: only update if non-empty and not masked
pwd_val = str(data.get("smtp_password", "")).strip()
if pwd_val and pwd_val != "••••••••••••":
settings["smtp_password"] = pwd_val
if "encryption" in data:
enc = str(data["encryption"]).lower().strip()
if enc in ["tls", "ssl", "none"]:
settings["encryption"] = enc
if "from_name" in data and str(data["from_name"]).strip():
settings["from_name"] = str(data["from_name"]).strip()
if "from_address" in data and str(data["from_address"]).strip():
settings["from_address"] = str(data["from_address"]).strip()
# Booleans / Toggles
def parse_bool(val, default=False):
if val is None:
return default
return str(val).lower() in ["1", "true", "on", "yes"]
is_form_submit = (handler.headers.get("Content-Type", "").startswith("application/x-www-form-urlencoded"))
if is_form_submit:
settings["require_email_verification"] = ("require_email_verification" in data)
settings["notify_welcome"] = ("notify_welcome" in data)
settings["notify_transactions"] = ("notify_transactions" in data)
settings["notify_security"] = ("notify_security" in data)
settings["notify_password_reset"] = ("notify_password_reset" in data)
settings["notify_deposits"] = ("notify_deposits" in data)
settings["notify_withdrawals"] = ("notify_withdrawals" in data)
settings["notify_cards"] = ("notify_cards" in data)
else:
for k in ["require_email_verification", "notify_welcome", "notify_transactions", "notify_security",
"notify_password_reset", "notify_deposits", "notify_withdrawals", "notify_cards"]:
if k in data:
settings[k] = parse_bool(data[k])
STATE["email_settings"] = settings
STATE.setdefault("audit_logs", []).append({
"action": "admin_email_settings_updated",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Updated SMTP and notification trigger configuration (Host: {settings['smtp_host']}:{settings['smtp_port']}, Enc: {settings['encryption']})"
})
if handler.headers.get("X-Requested-With") == "XMLHttpRequest" or "application/json" in handler.headers.get("Accept", ""):
handler.send_response(200)
handler.send_header("Content-Type", "application/json")
handler.end_headers()
safe_settings = dict(settings)
safe_settings["smtp_password"] = "••••••••••••"
handler.wfile.write(json.dumps({"success": True, "message": "Email settings saved successfully.", "settings": safe_settings}).encode('utf-8'))
return True
handler.send_response(302)
handler.send_header("Location", "/admin/settings/email?saved=1")
handler.end_headers()
return True
# 14. Admin Send Test Email
if path in ["/admin/settings/email/test", "/admin/email-settings/test"]:
recipient = str(data.get("recipient_email") or data.get("recipient") or STATE.get("user", {}).get("email", "admin@bnkfx.com")).strip()
settings = mail_service.get_email_settings(STATE)
success = mail_service.send_notification_email(
recipient=recipient,
template="test",
context={
"host": settings.get("smtp_host", "smtp.mailgun.org"),
"port": settings.get("smtp_port", 587),
"encryption": settings.get("encryption", "TLS"),
"sender": f"{settings.get('from_name', 'Alliance Bank')} <{settings.get('from_address', 'notifications@bnkfx.com')}>",
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S UTC")
},
user_id=1,
STATE=STATE
)
STATE.setdefault("audit_logs", []).append({
"action": "admin_email_test_dispatched",
"user_id": 1,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Dispatched SMTP diagnostic test email to {recipient} (Status: {'Sent' if success else 'Failed'})"
})
handler.send_response(200)
handler.send_header("Content-Type", "application/json")
handler.end_headers()
handler.wfile.write(json.dumps({
"success": success,
"message": f"SMTP diagnostic test email dispatched to {recipient}." if success else "Failed to dispatch test email.",
"recipient": recipient,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"log_id": len(STATE.get("email_logs", []))
}).encode('utf-8'))
return True
# 15. Platform Settings Update (Super Admin Only)
if path == "/admin/settings/platform":
actor_user = STATE.get("user", {})
if not platform_settings_service.is_super_admin(actor_user):
handler.send_response(403)
handler.send_header("Content-Type", "application/json")
handler.end_headers()
handler.wfile.write(json.dumps({
"success": False,
"error": "Access denied. Only Super Administrators can modify global platform settings."
}).encode('utf-8'))
return True
success, msg = platform_settings_service.update_platform_settings(data, STATE, actor_user)
active_settings = platform_settings_service.get_platform_settings(STATE)
if handler.headers.get("X-Requested-With") == "XMLHttpRequest" or "application/json" in handler.headers.get("Accept", ""):
handler.send_response(200 if success else 400)
handler.send_header("Content-Type", "application/json")
handler.end_headers()
handler.wfile.write(json.dumps({
"success": success,
"message": msg,
"settings": active_settings
}).encode('utf-8'))
return True
handler.send_response(302)
handler.send_header("Location", "/admin/settings/platform?saved=1")
handler.end_headers()
return True
# 16. Platform Asset / Logo Upload (Super Admin Only)
if path == "/admin/settings/upload-asset":
actor_user = STATE.get("user", {})
if not platform_settings_service.is_super_admin(actor_user):
handler.send_response(403)
handler.send_header("Content-Type", "application/json")
handler.end_headers()
handler.wfile.write(json.dumps({
"success": False,
"error": "Access denied. Super Administrator privileges required."
}).encode('utf-8'))
return True
asset_type = str(data.get("asset_type", "primary_logo")).strip()
asset_data = data.get("asset_data", "")
asset_mime = data.get("asset_mime", "image/png").lower()
allowed_mimes = ["image/png", "image/jpeg", "image/webp", "image/svg+xml", "image/x-icon", "image/vnd.microsoft.icon"]
if asset_mime not in allowed_mimes:
handler.send_response(400)
handler.send_header("Content-Type", "application/json")
handler.end_headers()
handler.wfile.write(json.dumps({
"success": False,
"error": f"Invalid asset format: {asset_mime}. Allowed types: PNG, JPEG, WebP, SVG, ICO."
}).encode('utf-8'))
return True
# Construct asset path / data URL
if asset_data:
new_url = f"/admin/uploads/platform-asset/{asset_type}"
STATE.setdefault("_platform_assets", {})[asset_type] = {
"data": asset_data,
"mime": asset_mime
}
else:
new_url = str(data.get("asset_url", f"/assets/images/logo/{asset_type}.svg"))
# Map to platform settings
target_field = f"{asset_type}_url" if not asset_type.endswith("_url") else asset_type
platform_settings_service.update_platform_settings({target_field: new_url}, STATE, actor_user)
handler.send_response(200)
handler.send_header("Content-Type", "application/json")
handler.end_headers()
handler.wfile.write(json.dumps({
"success": True,
"url": new_url,
"asset_type": asset_type,
"message": f"Asset '{asset_type}' updated successfully."
}).encode('utf-8'))
return True
return False
def handle_admin_get(handler, parsed, path, STATE, build_admin_layout):
# 0.1 Uploaded Platform Asset
if path.startswith("/admin/uploads/platform-asset/"):
try:
asset_type = path.split("/")[-1]
asset_rec = STATE.get("_platform_assets", {}).get(asset_type)
if asset_rec and asset_rec.get("data"):
import base64
raw = asset_rec["data"]
if isinstance(raw, bytes):
b = raw
else:
if "," in raw:
raw = raw.split(",", 1)[1]
b = base64.b64decode(raw)
handler.send_response(200)
handler.send_header("Content-Type", asset_rec.get("mime", "image/png"))
handler.send_header("Content-Length", str(len(b)))
handler.end_headers()
handler.wfile.write(b)
return True
except Exception:
pass
handler.send_response(404)
handler.end_headers()
return True
# 0. Uploaded Payment Method Icon
if path.startswith("/admin/uploads/icon/"):
try:
pm_id = int(path.split("/")[-1])
pm = next((p for p in STATE.get("payment_methods", []) if p["id"] == pm_id), None)
if pm and pm.get("icon_data"):
import base64
icon_raw = pm["icon_data"]
if isinstance(icon_raw, bytes):
icon_bytes = icon_raw
else:
if "," in icon_raw:
icon_raw = icon_raw.split(",", 1)[1]
icon_bytes = base64.b64decode(icon_raw)
handler.send_response(200)
handler.send_header("Content-Type", pm.get("icon_mime", "image/png"))
handler.send_header("Content-Length", str(len(icon_bytes)))
handler.end_headers()
handler.wfile.write(icon_bytes)
return True
except Exception:
pass
handler.send_response(404)
handler.end_headers()
return True
# 1. Exit Impersonation
if path == "/admin/impersonate/exit":
target_id = STATE.get("user", {}).get("id", 2)
STATE["impersonation"] = None
STATE["user"] = next((u for u in STATE.get("users", []) if u["id"] == 2), STATE["users"][0])
STATE.setdefault("audit_logs", []).append({
"action": "admin_user_impersonation_ended",
"user_id": 1,
"ip_address": "127.0.0.1",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"details": f"Support session ended for user #{target_id}"
})
handler.send_response(302)
handler.send_header("Location", f"/admin/users/{target_id}")
handler.end_headers()
return True
# 2. Payment Methods Index
if path == "/admin/payment-methods":
pms = STATE.get("payment_methods", [])
total_pms = len(pms)
active_pms = sum(1 for p in pms if p.get("status") == "active")
disabled_pms = sum(1 for p in pms if p.get("status") == "disabled")
archived_pms = sum(1 for p in pms if p.get("status") == "archived")
qs = urllib.parse.parse_qs(parsed.query)
search_q = qs.get("search", [""])[0].lower()
status_f = qs.get("status", [""])[0].lower()
filtered_pms = []
for p in pms:
if status_f and p.get("status") != status_f:
continue
if search_q:
if search_q not in p.get("name", "").lower() and search_q not in p.get("code", "").lower() and search_q not in p.get("wallet_address", "").lower():
continue
filtered_pms.append(p)
pm_rows = ""
for p in filtered_pms:
st = p.get("status", "active")
if st == "active":
st_badge = 'Archived '
addr = p.get("wallet_address", "")
addr_trunc = (addr[:10] + "..." + addr[-8:]) if len(addr) > 18 else addr
hist_count = sum(1 for h in STATE.get("wallet_address_history", []) if h.get("payment_method_id") == p["id"])
hist_btn = f'
{p['name']}
{p['code']} • Order #{p.get('sort_order', 1)}
{p['network']}
{addr_trunc}
${p['min_deposit']:.2f} - ${p['max_deposit']:,.0f} USD
{p['required_confirmations']} blocks • {p.get('processing_fee_pct', 0):.1f}% fee
{st_badge}
"""
body = f'''
Payment Methods & Deposit Wallets
Manage customer cryptocurrency deposit assets, network wallet addresses, deposit limits, and confirmations.
Add Payment Method
Total Configured
{total_pms}
Configured methods
Active for Deposits
{active_pms}
Live on customer app
Disabled Methods
{disabled_pms}
Temporarily suspended
Archived Methods
{archived_pms}
Decommissioned
Configured Deposit Assets
Changes immediately synchronized with customer app
Asset & Method
Network
Deposit Wallet Address
Limits (USD)
Confirmations / Fee
Status
Actions
{pm_rows}
Add Payment Method
Configure a new deposit method with asset, network, wallet address, icon, QR code, and limits.
Update configuration. Wallet address changes are logged in the audit trail.
'''
html = build_admin_layout(body, "Payment Methods & Wallets | Bnkfx Admin", path)
handler.wfile.write(html.encode('utf-8'))
return True
# 3. User Management Index
if path == "/admin/users":
users_list = STATE.get("users", [])
total_users = len(users_list)
active_users = sum(1 for u in users_list if u.get("status") == "active")
suspended_users = sum(1 for u in users_list if u.get("status") == "suspended")
email_verified_users = sum(1 for u in users_list if u.get("email_verified_at") is not None)
email_pending_users = sum(1 for u in users_list if u.get("email_verified_at") is None)
kyc_approved_users = sum(1 for u in users_list if u.get("kyc_status") == "approved")
kyc_pending_users = sum(1 for u in users_list if u.get("kyc_status") in ["pending_review", "under_review"])
kyc_rejected_users = sum(1 for u in users_list if u.get("kyc_status") == "rejected")
qs = urllib.parse.parse_qs(parsed.query)
search_q = qs.get("search", [""])[0].lower()
status_f = qs.get("status", [""])[0].lower()
kyc_f = qs.get("kyc_status", [""])[0].lower()
filtered_users = []
for u in users_list:
if status_f and u.get("status") != status_f:
continue
if kyc_f and u.get("kyc_status") != kyc_f:
continue
if search_q:
full_name = f"{u.get('first_name', '')} {u.get('last_name', '')}".lower()
email = u.get("email", "").lower()
uid_str = str(u.get("id", ""))
if search_q not in full_name and search_q not in email and search_q not in uid_str:
continue
filtered_users.append(u)
user_rows = ""
for u in filtered_users:
bal = STATE["accounts"][0]["balance"] if u["id"] == STATE["user"]["id"] else 0.00
st = u.get("status", "active")
st_badge = 'Unverified '
user_rows += f"""
{u['first_name'][0]}{u['last_name'][0]}
${bal:,.2f}
{st_badge}
{em_badge}
{kyc_badge}
{u.get('created_at', '2026-10-20')[:10]}
"""
body = f'''
Customer Accounts & User Management
Oversee customer accounts, enforce compliance, verify emails, review KYC, and perform support sessions.
Total: {total_users} Users
Total Users
{total_users}
Registered
Active
{active_users}
Operational
Suspended
{suspended_users}
Blocked
Email Verified
{email_verified_users}
Confirmed
Email Pending
{email_pending_users}
Unverified
KYC Approved
{kyc_approved_users}
Verified Tier
KYC Review
{kyc_pending_users}
Needs Review
KYC Rejected
{kyc_rejected_users}
Requires Docs
Customer Profile
USD Balance
Account Status
Email Verification
KYC Status
Joined Date
Actions
{user_rows}
'''
html = build_admin_layout(body, "Customer Accounts | Bnkfx Admin", path)
handler.wfile.write(html.encode('utf-8'))
return True
# 4. User Profile Show View
if path.startswith("/admin/users/"):
uid = int(path.split("/")[3])
u = next((x for x in STATE.get("users", []) if x["id"] == uid), STATE["user"])
qs = urllib.parse.parse_qs(parsed.query)
tab = qs.get("tab", ["profile"])[0]
bal = STATE["accounts"][0]["balance"] if u["id"] == STATE["user"]["id"] else 0.00
st = u.get("status", "active")
st_badge = '{tlabel}
Edit Customer Profile
Update contact details and address information.
'''
elif tab == "status":
action_btn = f'''
''' if u['status'] == "active" else f'''
'''
tab_body = f'''
Account Status: {u['status'].upper()}
Suspended accounts are strictly blocked from initiating deposits, sending transfers, or issuing cards.
{action_btn}
'''
elif tab == "email":
action_form = f'''
''' if not u.get("email_verified_at") else f'''
'''
tab_body = f'''
Email Verification
Current status: {("Verified on " + u['email_verified_at']) if u.get("email_verified_at") else "Pending Verification"}
{action_form}
'''
elif tab == "kyc":
tab_body = f'''
KYC Verification Tier: {u.get("kyc_status", "unverified").upper()}
Document Type: {u.get("kyc_document_type", "Passport")}
'''
elif tab == "balance":
tab_body = f'''
Account Balances & Double-Entry Ledger
Adjust Balance
Primary Checking (USD)
${bal:,.2f} USD
'''
elif tab == "activity":
tab_body = f'''
Transaction Activity
Total transactions recorded: {len(STATE.get("transfers", [])) + len(STATE.get("deposits", []))}
'''
elif tab == "cards":
tab_body = f'''
Issued Payment Cards
Customer currently has {len(STATE.get("cards", []))} payment cards.
'''
elif tab == "audit":
logs = [l for l in STATE.get("audit_logs", []) if str(u['id']) in str(l.get("details", ""))]
log_rows = "".join([f'{l["action"]} {l.get("details", "")} {l.get("created_at", "")} No audit logs for this user.
Security & Administrative Audit Trail
'''
body = f'''
{u['first_name'][0]}{u['last_name'][0]}
{u['first_name']} {u['last_name']}
{st_badge}
{em_badge}
{kyc_badge}
{u['email']} • {u.get('phone', 'No phone')} • {u.get('country', 'United States')}
Manage Balance
{tab_links}
{tab_body}
Adjust Customer Balance
Atomic double-entry ledger adjustment.
'''
html = build_admin_layout(body, f"User: {u['first_name']} {u['last_name']} | Bnkfx Admin", path)
handler.wfile.write(html.encode('utf-8'))
return True
# 5. KYC Document Viewer
if path.startswith("/admin/kyc/document/"):
doc_id = path.split("/")[4]
sub = next((k for k in STATE.get("kyc_verifications", []) if str(k["id"]) == doc_id), None)
doc_type = sub.get("document_type", "Government ID") if sub else "Government ID"
doc_user = f"{sub.get('first_name', '')} {sub.get('last_name', '')}" if sub else "Customer"
svg = f'''
{doc_type}
Holder: {doc_user}
Document ID: {sub.get("id_number", "VERIFIED-DOC") if sub else "DL-98234-UK"}
OFFICIAL DIGITAL SECURITY COPY
P<USA{sub.get("last_name", "SMITH").upper()}<<{sub.get("first_name", "JOHN").upper()}<<<<<<<<<<<<<<<<<<<<<<
98234UK5829104M2801014<<<<<<<<<<<<<<04
Alliance Bank Compliance Cryptographic Proof • Confidential Verification Document
'''
handler.send_response(200)
handler.send_header("Content-Type", "image/svg+xml")
handler.end_headers()
handler.wfile.write(svg.encode('utf-8'))
return True
# 6. KYC Compliance Management Index
if path == "/admin/kyc":
subs = STATE.get("kyc_verifications", [])
total_sub = len(subs)
pending = sum(1 for s in subs if s.get("status") == "pending_review")
under_review = sum(1 for s in subs if s.get("status") == "under_review")
approved = sum(1 for s in subs if s.get("status") == "approved")
rejected = sum(1 for s in subs if s.get("status") == "rejected")
add_info = sum(1 for s in subs if s.get("status") == "additional_info_required")
qs = urllib.parse.parse_qs(parsed.query)
search_q = qs.get("search", [""])[0].lower().strip()
status_filter = qs.get("status", [""])[0].strip()
doc_filter = qs.get("document_type", [""])[0].strip()
filtered_subs = subs
if status_filter:
filtered_subs = [s for s in filtered_subs if s.get("status") == status_filter]
if doc_filter:
filtered_subs = [s for s in filtered_subs if s.get("document_type") == doc_filter]
if search_q:
filtered_subs = [s for s in filtered_subs if (
search_q in s.get("first_name", "").lower() or
search_q in s.get("last_name", "").lower() or
search_q in s.get("email", "").lower() or
search_q in s.get("account_number", "").lower() or
search_q in str(s.get("user_id", "")).lower()
)]
rows_html = ""
for s in filtered_subs:
st = s.get("status", "pending_review")
badge_class = {
"approved": "bg-[#20C997]/10 text-[#20C997] border-[#20C997]/20",
"pending_review": "bg-[#F5B942]/10 text-[#F5B942] border-[#F5B942]/20",
"under_review": "bg-[#246BFD]/10 text-[#246BFD] border-[#246BFD]/20",
"rejected": "bg-[#F05D6C]/10 text-[#F05D6C] border-[#F05D6C]/20",
"additional_info_required": "bg-orange-500/10 text-orange-400 border-orange-500/20"
}.get(st, "bg-white/10 text-[#8F9CAE] border-white/10")
badge_label = {
"approved": "Approved",
"pending_review": "Pending Review",
"under_review": "Under Review",
"rejected": "Rejected",
"additional_info_required": "Info Required"
}.get(st, st.capitalize())
s_json = json.dumps(s).replace("'", "'")
rows_html += f'''
{s.get("first_name", "")} {s.get("last_name", "")}
{s.get("email", "")} • ID #{s.get("user_id", "")}
{s.get("account_number", "—")}
{s.get("document_type", "ID")}
{s.get("id_number", "—")}
{s.get("submitted_at", "Just now")}
{badge_label}
{s.get("updated_at", "—")}
Review Submission
'''
if not rows_html:
rows_html = '''
No KYC Submissions Found
There are no verification documents matching your criteria.
Compliance & Regulatory
•
Tier 2 Verification
Authoritative identity verification queue, document inspection, and review decisions.
Total Submissions
{total_sub}
Under Review
{under_review}
Info Requested
{add_info}
Customer
Account Number
Document Type
Submission Date
KYC Status
Last Updated
Review Action
{rows_html}
Document Type
Document ID Number
Submission Date
Current Status
'''
html = build_admin_layout(body, "KYC Compliance Management | Bnkfx Admin", path)
handler.wfile.write(html.encode('utf-8'))
return True
# 4. Stock Catalog Management
if path == "/admin/stocks":
stocks = STATE.get("stocks", [])
plans = STATE.get("investment_plans", [])
total_stocks = len(stocks)
active_stocks = sum(1 for s in stocks if s.get("status") == "active")
total_invested = sum(float(i.get("amount", 0)) for i in STATE.get("investments", []))
total_redeemed = sum(float(i.get("expected_payout", 0)) for i in STATE.get("investments", []) if i.get("status") == "redeemed")
stocks_rows_html = ""
for stk in stocks:
plan = next((p for p in plans if p["id"] == stk.get("default_plan_id")), None)
plan_name = plan["name"] if plan else "Custom"
ret_pct = plan["return_percentage"] if plan else 20.0
dur_hrs = plan["duration_hours"] if plan else 48
is_active = stk.get("status") == "active"
stk_json = json.dumps(stk).replace('"', '"')
stocks_rows_html += f'''
{stk['ticker']}
STOCK
{stk['name']}
{stk.get('category', 'Technology')}
{plan_name}
+{ret_pct:g}%
{dur_hrs} hrs duration
{stk.get('status', 'active')}
{stk.get('display_order', 0)}
'''
plan_options_html = "".join([f'{p["name"]} (+{p["return_percentage"]:g}%, {p["duration_hours"]}h) ' for p in plans])
body = f'''
Stock Catalog & Assets
Manage institutional stock offerings, default plan assignments, and visibility in customer dashboard.
Total Stock Assets
{total_stocks}
Active in Catalog
{active_stocks}
Total Capital Allocated
${total_invested:,.2f}
Total Redemptions
${total_redeemed:,.2f}
Stock Asset
Category
Default Plan
Contractual Terms
Status
Display Order
Actions
{stocks_rows_html if stocks_rows_html else 'No stocks registered.
'''
html = build_admin_layout(body, "Stock Management & Investment Products | Bnkfx Admin", path)
handler.wfile.write(html.encode('utf-8'))
return True
# 5. Investment Plans Management
if path == "/admin/plans":
plans = STATE.get("investment_plans", [])
plans_cards_html = ""
for p in plans:
is_active = p.get("status") == "active"
is_rec = bool(p.get("is_recommended"))
is_unl = bool(p.get("is_unlimited") or p.get("max_amount") is None)
min_amt = float(p.get("min_amount", 50))
max_amt = p.get("max_amount")
max_str = "Unlimited" if is_unl else f"${float(max_amt):,.2f}"
plans_cards_html += f'''
{f'
Recommended ' if is_rec else ''}
{p['name']}
{p.get('slug', '')}
{p.get('status', 'active')}
Contractual Return
{p['duration_hours']} Hours Lock Period
Min Allocation:
${min_amt:,.2f}
Max Allocation:
{max_str}
Edit Tier
Investment Plans & Yield Tiers
Configure minimum/maximum allocation thresholds, lock durations, and fixed contractual return rates.
{plans_cards_html}
'''
html = build_admin_layout(body, "Investment Plans & Yield Tiers | Bnkfx Admin", path)
handler.wfile.write(html.encode('utf-8'))
return True
# 6. User Investments Oversight
if path == "/admin/investments":
investments = STATE.get("investments", [])
total_vol = sum(float(i.get("amount", 0)) for i in investments)
active_vol = sum(float(i.get("amount", 0)) for i in investments if i.get("status") == "active")
active_cnt = sum(1 for i in investments if i.get("status") == "active")
matured_payout = sum(float(i.get("expected_payout", 0)) for i in investments if i.get("status") == "matured")
matured_cnt = sum(1 for i in investments if i.get("status") == "matured")
redeemed_vol = sum(float(i.get("expected_payout", 0)) for i in investments if i.get("status") == "redeemed")
inv_rows_html = ""
for inv in investments:
st = str(inv.get("status", "active")).lower()
badge_class = "bg-emerald-500/10 text-emerald-400 border border-emerald-500/20" if st == "matured" else ("bg-purple-500/10 text-purple-400 border border-purple-500/20" if st == "redeemed" else "bg-blue-500/10 text-blue-400 border border-blue-500/20")
user_rec = STATE.get("user", {})
user_name = f"{user_rec.get('first_name', 'Customer')} {user_rec.get('last_name', '')}"
user_email = user_rec.get('email', 'customer@bnkfx.com')
inv_rows_html += f'''
{inv.get('reference', 'INV')}
{user_name}
{user_email}
{inv.get('stock_ticker')} · {inv.get('stock_name')}
{inv.get('plan_name')} (+{inv.get('return_percentage', 0):g}%)
${float(inv.get('amount', 0)):,.2f}
${float(inv.get('expected_payout', 0)):,.2f}
{inv.get('maturity_at', '')}
{st}
{'Redeemed ' + str(inv.get('redeemed_at', '')) if st == 'redeemed' else 'Active Cycle'}
'''
body = f'''
User Investment Oversight & Redemptions
Audit all platform customer investment allocations, lock periods, maturities, and payout redemptions.
Total Platform Volume
${total_vol:,.2f}
Active Capital (Locked)
${active_vol:,.2f}
{active_cnt} active positions
Matured (Unredeemed)
${matured_payout:,.2f}
{matured_cnt} matured positions
Total Redeemed Paid
${redeemed_vol:,.2f}
Reference
Customer
Stock & Plan
Invested Amount
Expected Payout
Maturity Date
Status
Audit Details
{inv_rows_html if inv_rows_html else 'No investment records found.
'''
html = build_admin_layout(body, "User Investment Oversight & Redemptions | Bnkfx Admin", path)
handler.wfile.write(html.encode('utf-8'))
return True
# 10. Card Products & Applications Administration
if path == "/admin/cards":
qs = urllib.parse.parse_qs(parsed.query)
active_tab = qs.get("tab", ["products"])[0]
curr_status = qs.get("status", ["all"])[0].lower()
search_q = qs.get("search", [""])[0].lower()
products = STATE.get("card_products", [])
orders = STATE.get("card_orders", [])
tot_prods = len(products)
act_prods = sum(1 for p in products if p.get("is_active", 1) == 1)
inact_prods = tot_prods - act_prods
tot_apps = len(orders)
pend_apps = sum(1 for o in orders if o.get("status") in ["pending", "processing"])
appr_apps = sum(1 for o in orders if o.get("status") in ["completed", "approved", "active"])
rej_apps = sum(1 for o in orders if o.get("status") == "rejected")
# Build Products Cards HTML
prod_cards_html = ""
for p in products:
is_act = (p.get("is_active", 1) == 1)
design = p.get("design", p.get("card_design", "obsidian"))
if design == "midnight_blue":
bg_grad = "from-[#0B2545] via-[#133E7C] to-[#1E3A8A] border-blue-400/30"
elif design == "titanium_gold":
bg_grad = "from-[#3A2207] via-[#854D0E] to-[#D97706] border-amber-400/30"
elif design == "crypto_black":
bg_grad = "from-[#000000] via-[#111827] to-[#1F2937] border-purple-500/30"
else:
bg_grad = "from-[#0F1318] via-[#161B22] to-[#21262D] border-gray-700/50"
p_fee = float(p.get("fee", p.get("issuance_fee", 0.0)))
p_min_bal = float(p.get("min_balance_required", 0.0))
p_daily_lim = float(p.get("limit", p.get("daily_limit", 5000.0)))
p_monthly_lim = float(p.get("monthly_limit", 25000.0))
feats_list = p.get("features", [])
feats_html = "".join([f'{f} +{len(feats_list) - 3} more benefits '
p_json_str = html_lib.escape(json.dumps(p), quote=True)
prod_cards_html += f'''
{p['name']}
•••• •••• •••• 0000
Card Type
{p.get('type', 'virtual').upper()}
{p.get('design', 'obsidian').upper()}
{p['name']}
{p.get('type', 'virtual')} Card • {p.get('network', 'Visa')}
{'● Active' if is_act else '○ Disabled'}
Issuance Fee
{'$' + f'{p_fee:.2f}' if p_fee > 0 else 'Free ($0.00)'}
Min Balance Req.
{'$' + f'{p_min_bal:.2f}' if p_min_bal > 0 else 'None ($0.00)'}
Daily Spend Limit
${p_daily_lim:,.2f}
Monthly Limit
${p_monthly_lim:,.2f}
'''
# Filter Applications
filtered_orders = []
for o in orders:
o_st = str(o.get("status", "pending")).lower()
if curr_status == "pending" and o_st not in ["pending", "processing"]:
continue
elif curr_status in ["completed", "approved"] and o_st not in ["completed", "approved", "active"]:
continue
elif curr_status == "rejected" and o_st != "rejected":
continue
if search_q:
haystack = f"{o.get('reference', '')} {o.get('user_name', '')} {o.get('email', '')} {o.get('product_name', '')}".lower()
if search_q not in haystack:
continue
filtered_orders.append(o)
app_rows_html = ""
for o in filtered_orders:
o_st = str(o.get("status", "pending")).lower()
is_pending = o_st in ["pending", "processing"]
is_approved = o_st in ["completed", "approved", "active"]
is_rejected = (o_st == "rejected")
o_fee = float(o.get("payment_amount", 0.0))
o_min_bal = float(o.get("min_balance_required", 0.0))
o_ref = o.get("reference", f"ORD-CRD-{o['id']}")
if is_pending:
st_pill = '
'''
elif is_approved:
st_pill = 'Live on Account '
elif is_rejected:
rej_reason = html_lib.escape(str(o.get("rejection_reason", "Compliance criteria not met")))
st_pill = f'''
"{rej_reason}"
'''
actions_html = 'Refund Settled '
else:
st_pill = f'{o_st} '
actions_html = '—'
app_rows_html += f'''
{o_ref}
{o.get('user_name', 'Abraham Elom')}
{o.get('email', '—')}
{o.get('product_name', 'Virtual Card')}
{o.get('card_type', 'virtual')} • {o.get('network', 'Visa')}
{f'${o_fee:.2f}' if o_fee > 0 else 'Free'}
{f'${o_min_bal:.2f}' if o_min_bal > 0 else '$0.00'}
{o.get('created_at', time.strftime('%Y-%m-%d %H:%M'))}
{st_pill}
{actions_html}
'''
body = f'''
Virtual & Physical Card Management
Configure card products, manage issuance limits & eligibility requirements, and process customer card applications.
Add Card Product
Total Products
{tot_prods}
Configured tiers
Active Catalog
{act_prods}
Available to users
Suspended
{inact_prods}
Hidden from catalog
Applications
{tot_apps}
All submissions
Pending Review
{pend_apps}
Awaiting action
{'
' if pend_apps > 0 else ''}
Issued / Active
{appr_apps}
Live customer cards
Rejected
{rej_apps}
Fees refunded
Card Products Catalog
{tot_prods}
Card Applications
{f'{pend_apps} ' if pend_apps > 0 else f'{tot_apps} '}
{prod_cards_html if prod_cards_html else '
No card products configured.
'}
Order Ref
Applicant / Customer
Card Product
Fee Paid
Req. Balance
Date
Status
Actions
{app_rows_html if app_rows_html else '
No card applications found
Customer submissions will appear here for review and issuance.
Create Card Product
Configure catalog tier, fee, eligibility minimum, and limits.
Edit Card Product
Modify parameters and fees for this card product.
Reject Card Application
Document the rejection reason. Any issuance fee paid will be refunded to customer balance.
'''
html = build_admin_layout(body, "Card Marketplace Administration | Bnkfx Admin", path)
handler.wfile.write(html.encode('utf-8'))
return True
# 10. Email Settings Admin View
if path in ["/admin/settings/email", "/admin/email-settings"]:
settings = mail_service.get_email_settings(STATE)
email_logs = STATE.get("email_logs", [])
total_emails = len(email_logs)
sent_emails = sum(1 for e in email_logs if e.get("status") in ["sent", "delivered"])
failed_emails = sum(1 for e in email_logs if e.get("status") == "failed")
success_pct = f"{(sent_emails / total_emails * 100):.1f}%" if total_emails > 0 else "100.0%"
triggers = [
("notify_welcome", "Customer Welcome Notifications", "Sent immediately upon customer registration with account details.", settings.get("notify_welcome", True)),
("notify_transactions", "Financial Transaction Confirmations", "Sent when money transfers, exchanges, or fee settlements finalize.", settings.get("notify_transactions", True)),
("notify_security", "Security & Access Alerts", "Sent on logins, password modifications, and security setting updates.", settings.get("notify_security", True)),
("notify_password_reset", "Password Reset Recovery Links", "Sent when users request password recovery with single-use tokens.", settings.get("notify_password_reset", True)),
("notify_deposits", "Crypto & Fiat Deposit Updates", "Sent when deposits are initiated, confirmed, credited, or rejected.", settings.get("notify_deposits", True)),
("notify_withdrawals", "Withdrawal Status Updates", "Sent when customer withdrawal disbursements process.", settings.get("notify_withdrawals", True)),
("notify_cards", "Virtual Card Application & Issuance", "Sent when card applications are submitted, approved, or declined.", settings.get("notify_cards", True)),
]
active_triggers_count = sum(1 for _, _, _, enabled in triggers if enabled)
query = urllib.parse.parse_qs(parsed.query)
saved_banner = ""
if "saved" in query:
saved_banner = '''
Settings Persisted Successfully
SMTP credentials and notification policies have been committed to the runtime environment.
'''
triggers_html = ""
for field_name, title_text, desc_text, is_enabled in triggers:
checked = "checked" if is_enabled else ""
triggers_html += f'''
'''
require_verif_checked = "checked" if settings.get("require_email_verification", True) else ""
body = f'''
Admin
Email Settings & SMTP Configuration
Email System Management
Configure SMTP delivery gateways, credentials, security verification rules, and automated transactional events.
{saved_banner}
Total Emails Dispatched
{total_emails}
{sent_emails} sent successfully
Delivery Reliability
{success_pct}
{failed_emails} reported delivery failures
Active Triggers
{active_triggers_count} / {len(triggers)} enabled
Automated event dispatches
Verification Policy
{'Enforced ' if settings.get("require_email_verification", True) else 'Optional '}
Requires email activation
SMTP Diagnostic Tool
Send a live test email to verify gateway connectivity.
Diagnostic Monitor
Ready
[SYS] Mail gateway idle. Ready to test socket relay.
Security Architecture
Financial decoupling & credential protection.
Financial Decoupling: Transaction ledger updates are executed first. Mail delivery exceptions never reverse completed monetary transfers.
Deduplication Guard: Unique idempotency keys prevent duplicate email dispatch on retries or repeated webhook triggers.
Data Masking: Sensitive authentication passwords and full PAN card numbers are strictly stripped from outgoing templates.
Audit Trail: All SMTP credential modifications and test events are timestamped in the administrative security audit log.
'''
html = build_admin_layout(body, "Email Settings | Bnkfx Admin", path)
handler.wfile.write(html.encode('utf-8'))
return True
# 11. Email Delivery Logs Admin View
if path == "/admin/email-logs":
email_logs = list(STATE.get("email_logs", []))
email_logs.reverse() # Newest first
total_count = len(email_logs)
sent_count = sum(1 for e in email_logs if e.get("status") in ["sent", "delivered"])
failed_count = sum(1 for e in email_logs if e.get("status") == "failed")
queued_count = sum(1 for e in email_logs if e.get("status") == "queued")
query = urllib.parse.parse_qs(parsed.query)
curr_status = query.get("status", ["all"])[0]
curr_template = query.get("template", ["all"])[0]
search_q = query.get("q", [""])[0].lower().strip()
filtered_logs = email_logs
if curr_status != "all":
if curr_status == "sent":
filtered_logs = [e for e in filtered_logs if e.get("status") in ["sent", "delivered"]]
else:
filtered_logs = [e for e in filtered_logs if e.get("status") == curr_status]
if curr_template != "all":
filtered_logs = [e for e in filtered_logs if e.get("template") == curr_template]
if search_q:
filtered_logs = [e for e in filtered_logs if search_q in e.get("recipient", "").lower() or search_q in e.get("subject", "").lower() or search_q in e.get("template", "").lower()]
rows_html = ""
for log in filtered_logs:
log_id = log.get("id", 0)
status = log.get("status", "sent")
template = log.get("template", "default")
recipient = html_lib.escape(log.get("recipient", "N/A"))
subject = html_lib.escape(log.get("subject", "Notification"))
created_at = log.get("created_at", "N/A")
error_msg = log.get("error_message") or ""
if status in ["sent", "delivered"]:
status_badge = '{template.replace("_", " ")} '
# Metadata JSON string for inspection
meta_json = html_lib.escape(json.dumps(log.get("metadata", {}), indent=2))
error_display = f'{html_lib.escape(error_msg)}
' if error_msg else ''
rows_html += f'''
#{log_id}
{created_at}
{recipient}
{f'User ID: {log.get("user_id")}
' if log.get("user_id") else ''}
{subject}
{error_display}
{template_badge}
{status_badge}
'''
# JSON lookup dictionary for inspection modal
logs_json_lookup = html_lib.escape(json.dumps({l["id"]: l for l in email_logs}))
body = f'''
Email Delivery Logs
Audit outgoing transactional email payloads, transmission timestamps, and gateway error messages.
Total Logs Recorded
{total_count}
Audit log entries
Delivered Successfully
{sent_count}
Sent without error
Delivery Failures
{failed_count}
Requires inspection
Queued In Pipeline
{queued_count}
Awaiting relay
Log ID
Timestamp
Recipient
Subject
Template
Delivery Status
Details
{rows_html if rows_html else 'No email delivery logs match the selected filter criteria.
Recipient:
Template:
Status:
Recorded At:
Close
'''
html = build_admin_layout(body, "Email Delivery Logs | Bnkfx Admin", path)
handler.wfile.write(html.encode('utf-8'))
return True
# 12. Platform Settings Admin View (Super Admin Only)
if path == "/admin/settings/platform":
actor_user = STATE.get("user", {})
is_sa = platform_settings_service.is_super_admin(actor_user)
settings = platform_settings_service.get_platform_settings(STATE)
query = urllib.parse.parse_qs(parsed.query)
active_tab = query.get("tab", ["general"])[0]
saved_banner = ""
if "saved" in query:
saved_banner = '''
Platform Configuration Persisted
Global brand identity and platform settings have been committed and propagated.
'''
if not is_sa:
body = f'''
Super Administrator Access Required
Global platform settings, corporate branding, visual tokens, and server-side maintenance mode can only be modified by accounts holding the Super Administrator role.
Current Identity: {html_lib.escape(actor_user.get("email", "unknown"))} (Role: {html_lib.escape(actor_user.get("role", "admin"))})
'''
html = build_admin_layout(body, "Access Denied | Platform Settings", path)
handler.wfile.write(html.encode('utf-8'))
return True
# Render Full 10-Tab Interface for Super Admin
tab_list = [
("general", "General Settings", "fa-gear"),
("brand", "Brand Identity & Colors", "fa-palette"),
("contact", "Contact & Support", "fa-address-book"),
("social", "Social Media", "fa-share-nodes"),
("auth", "Auth & Registration", "fa-user-shield"),
("email", "Email Branding", "fa-envelope-open-text"),
("localization", "Localization", "fa-globe"),
("seo", "SEO & Metadata", "fa-magnifying-glass"),
("security", "Security & Access", "fa-lock"),
("legal", "Legal Pages", "fa-scale-balanced")
]
tabs_nav_html = ""
for t_id, t_name, t_icon in tab_list:
is_active = (t_id == active_tab)
classes = "bg-[#246BFD] text-white shadow-lg shadow-blue-600/30" if is_active else "text-[#8F9CAE] hover:text-white hover:bg-[#171A21]"
tabs_nav_html += f'''
{t_name}
'''
social_links = settings.get("social_links", {})
body = f'''
Admin
Platform Settings & Brand Identity
Platform Settings
Manage global website identity, logos, dynamic brand tokens, contact details, and maintenance controls.
{saved_banner}
Configuration Groups
{tabs_nav_html}
'''
html = build_admin_layout(body, "Platform Settings | Bnkfx Admin", path)
handler.wfile.write(html.encode('utf-8'))
return True
return False
![{p['code']}]({p.get('icon_url', '/assets/images/crypto/usdc.svg')})
![{stk['ticker']}]({stk.get('logo_url') or '/assets/images/stocks/' + stk['ticker'].lower() + '.svg'})