import os
import re
import json
import time
import base64
from pathlib import Path
from datetime import datetime, timezone
from urllib.parse import quote, unquote
import httpx
from flask import Flask, jsonify, render_template_string, request, session, redirect, url_for
BASE_DIR = Path(__file__).resolve().parent
# This is the SAME file oxide_configs_store.json that the Telegram bot
# (OXIDE_CORE) writes to when an admin sends a purchase's main sub link.
# Site and bot must share this file (same host/folder, or set the env var
# below to point at the shared path).
CONFIGS_STORE_FILE = Path(os.getenv("OXIDE_CONFIGS_STORE_FILE", BASE_DIR / "oxide_configs_store.json"))
# ---------------------------------------------------------------------------
# TODO: fill in the REAL remark/name of each config type exactly as it
# appears inside the raw main sub link (case-insensitive substring match is
# used, so partial names like "Gaming2" work fine).
# "secure" is the free-coin config - tell me its real name and I'll drop it
# in here.
# ---------------------------------------------------------------------------
CONFIG_TYPE_SOURCE_REMARK = {
"gaming": "Gaming2",
"secure": "REPLACE_WITH_REAL_COIN_CONFIG_NAME",
}
# The clean display name the config gets renamed to before being handed
# to the customer / V2rayNG.
CONFIG_TYPE_DISPLAY_NAME = {
"gaming": "Gaming",
"secure": "Coin",
}
CONFIG_TYPE_LABEL_FA = {
"gaming": "🎮 گیمینگ",
"secure": "🪙 سکه رایگان",
}
app = Flask(__name__)
app.secret_key = os.getenv("OXIDE_SECRET_KEY", "CHANGE_THIS_TO_A_RANDOM_SECRET")
app.config["PERMANENT_SESSION_LIFETIME"] = 86400
# =========================================================
# STORE ACCESS
# =========================================================
def load_configs_store():
try:
with CONFIGS_STORE_FILE.open("r", encoding="utf-8") as f:
data = json.load(f)
return data if isinstance(data, dict) else {"purchases": {}}
except (OSError, json.JSONDecodeError) as exc:
print(f"[oxide] could not read {CONFIGS_STORE_FILE}: {exc}", flush=True)
return {"purchases": {}}
def find_purchase(code):
code = str(code).strip()
if len(code) != 6 or not code.isdigit():
return None
store = load_configs_store()
return store.get("purchases", {}).get(code)
# =========================================================
# SUB PARSING / SINGLE-CONFIG EXTRACTION
# =========================================================
def decode_sub_body(raw_bytes):
text = raw_bytes.decode("utf-8", "ignore").strip()
try:
padded = text + "=" * (-len(text) % 4)
decoded = base64.b64decode(padded).decode("utf-8", "ignore")
if "://" in decoded:
return decoded
except Exception:
pass
return text
def get_config_remark(uri):
if uri.startswith("vmess://"):
try:
payload = uri[len("vmess://"):]
padded = payload + "=" * (-len(payload) % 4)
obj = json.loads(base64.b64decode(padded).decode("utf-8", "ignore"))
return str(obj.get("ps", ""))
except Exception:
return ""
if "#" in uri:
return unquote(uri.split("#", 1)[1])
return ""
def set_config_remark(uri, new_name):
if uri.startswith("vmess://"):
try:
payload = uri[len("vmess://"):]
padded = payload + "=" * (-len(payload) % 4)
obj = json.loads(base64.b64decode(padded).decode("utf-8", "ignore"))
obj["ps"] = new_name
new_payload = base64.b64encode(
json.dumps(obj, ensure_ascii=False).encode("utf-8")
).decode("utf-8")
return "vmess://" + new_payload
except Exception:
return uri
if "#" in uri:
base = uri.split("#", 1)[0]
return f"{base}#{quote(new_name)}"
return f"{uri}#{quote(new_name)}"
def fetch_personal_config(sub_link, config_type):
"""Downloads the main sub link (which contains many configs) and
returns just the ONE config matching this purchase's type, renamed
to a clean display name, plus the raw response headers (used for the
usage/expiry stats already exposed by the sub server)."""
with httpx.Client(
timeout=15,
follow_redirects=True,
headers={"User-Agent": "Mozilla/5.0 OXIDE-Nexus/1.0", "Accept": "*/*"},
) as client:
response = client.get(sub_link)
response.raise_for_status()
body = decode_sub_body(response.content)
lines = [ln.strip() for ln in body.splitlines() if ln.strip() and "://" in ln]
target_remark = CONFIG_TYPE_SOURCE_REMARK.get(config_type, "")
matched = None
if target_remark:
for line in lines:
if target_remark.lower() in get_config_remark(line).lower():
matched = line
break
if matched is None and lines:
matched = lines[0]
if matched is None:
return None, response.headers
display_name = CONFIG_TYPE_DISPLAY_NAME.get(config_type, "OXIDE CORE")
return set_config_remark(matched, display_name), response.headers
def parse_userinfo(value):
if not value:
return {}
result = {}
for part in value.replace(",", ";").split(";"):
if "=" in part:
key, val = part.split("=", 1)
result[key.strip().lower()] = val.strip()
return result
def integer(value):
try:
return int(float(value))
except (ValueError, TypeError):
return 0
def human_bytes(value):
value = max(0, value)
units = ["B", "KB", "MB", "GB", "TB", "PB"]
for unit in units:
if value < 1024 or unit == "PB":
return f"{int(value)} B" if unit == "B" else f"{value:.2f} {unit}"
value /= 1024
return "0 B"
def build_stats(headers):
header = (
headers.get("subscription-userinfo")
or headers.get("Subscription-Userinfo")
or headers.get("subscription-user-info")
)
info = parse_userinfo(header)
upload = integer(info.get("upload"))
download = integer(info.get("download"))
total = integer(info.get("total"))
expire = integer(info.get("expire"))
used = upload + download
remaining = max(total - used, 0) if total else 0
percent = min((used / total) * 100, 100) if total else 0
now = int(time.time())
if expire:
seconds = max(expire - now, 0)
days = (seconds + 86399) // 86400
status = "active" if expire > now else "expired"
expire_date = datetime.fromtimestamp(expire, timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
else:
days = None
status = "active"
expire_date = "Unlimited"
return {
"status": status,
"upload": human_bytes(upload),
"download": human_bytes(download),
"used": human_bytes(used),
"remaining": human_bytes(remaining),
"total": human_bytes(total),
"percent": round(percent, 1),
"days": days,
"expire": expire_date,
}
@app.after_request
def security_headers(response):
response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
response.headers["Pragma"] = "no-cache"
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["Referrer-Policy"] = "same-origin"
return response
# =========================================================
# ROUTES
# =========================================================
@app.route("/login", methods=["GET", "POST"])
def login():
if session.get("verified") and session.get("purchase_code"):
return redirect(url_for("home"))
error = None
if request.method == "POST":
code = request.form.get("code", "").strip()
purchase = find_purchase(code)
if purchase:
session.clear()
session.permanent = True
session["verified"] = True
session["purchase_code"] = str(code)
session["verified_at"] = int(time.time())
return redirect(url_for("home"))
error = "کد وارد شده معتبر نیست."
return render_template_string(LOGIN_HTML, error=error)
@app.route("/logout")
def logout():
session.clear()
return redirect(url_for("login"))
@app.route("/debug")
def debug_store():
# Temporary diagnostic route. Remove once everything works.
info = {
"script_folder (BASE_DIR)": str(BASE_DIR),
"looking_for_file_at": str(CONFIGS_STORE_FILE),
"file_exists": CONFIGS_STORE_FILE.exists(),
}
if CONFIGS_STORE_FILE.exists():
try:
info["file_readable"] = True
info["raw_content"] = CONFIGS_STORE_FILE.read_text(encoding="utf-8")
info["parsed_purchase_codes"] = list(load_configs_store().get("purchases", {}).keys())
except Exception as exc:
info["file_readable"] = False
info["read_error"] = str(exc)
return jsonify(info)
@app.route("/")
def home():
if not session.get("verified"):
return redirect(url_for("login"))
purchase = find_purchase(session.get("purchase_code")) or {}
config_type = purchase.get("config_type", "gaming")
return render_template_string(
PANEL_HTML,
config_type_label=CONFIG_TYPE_LABEL_FA.get(config_type, "کانفیگ شخصی"),
config_name=purchase.get("config_name") or CONFIG_TYPE_DISPLAY_NAME.get(config_type, "OXIDE CORE"),
)
@app.route("/api/info")
def api_info():
if not session.get("verified"):
return jsonify({"error": "Unauthorized"}), 401
purchase = find_purchase(session.get("purchase_code"))
if not purchase:
return jsonify({"error": "این خرید دیگر در سیستم موجود نیست."}), 404
try:
_, headers = fetch_personal_config(purchase["sub_link"], purchase.get("config_type", "gaming"))
result = build_stats(headers)
result["config_type_label"] = CONFIG_TYPE_LABEL_FA.get(purchase.get("config_type"), "-")
result["config_name"] = purchase.get("config_name") or CONFIG_TYPE_DISPLAY_NAME.get(purchase.get("config_type"))
return jsonify(result)
except Exception as exc:
return jsonify({"error": f"خطا در اتصال به سرور اشتراک: {exc}"}), 502
@app.route("/api/v2rayng")
def api_v2rayng():
if not session.get("verified"):
return jsonify({"error": "Unauthorized"}), 401
purchase = find_purchase(session.get("purchase_code"))
if not purchase:
return jsonify({"error": "این خرید دیگر در سیستم موجود نیست."}), 404
try:
renamed_uri, _ = fetch_personal_config(purchase["sub_link"], purchase.get("config_type", "gaming"))
if not renamed_uri:
return jsonify({"error": "کانفیگ مربوط به این خرید پیدا نشد."}), 502
deep_link = "v2rayng://install-config?url=" + quote(renamed_uri, safe="")
return jsonify({"deep_link": deep_link, "raw_config": renamed_uri})
except Exception as exc:
return jsonify({"error": f"خطا در اتصال به سرور اشتراک: {exc}"}), 502
LOGIN_HTML = r"""
OXIDE NEXUS • Access
OXIDE NEXUS
ورود امن
کد ۶ رقمی تایید خریدت رو وارد کن تا کانفیگ مخصوص خودت رو ببینی.