from flask import Flask, request, render_template_string, send_file, abort, redirect, url_for, Response
import os
import subprocess
import tempfile
from datetime import datetime
import shutil
import json
import uuid
from functools import wraps
app = Flask(__name__)
TOKEN_DB_PATH = "/opt/wgconfig/tokens.json"
SUBNET_DB_PATH = "/opt/wgconfig/subnets.json"
USED_LOG_PATH = "/opt/wgconfig/used_tokens.log"
PEERS_CONF_PATH = "/opt/wgconfig/wg0-peers.conf"
KEY_LOG_PATH = "/var/log/wg_keys_issued.log"
BACKUP_DIR = "/opt/wgconfig/backups"
SERVER_PUBLIC_KEY = "bsped9nemtP9fRfcVLoLL3JWTj7zHFYWJebAhgH7RFw="
SERVER_ENDPOINT = "24.249.187.133:51820"
ADMIN_USERNAME = "admin"
ADMIN_PASSWORD = "changeme"
def check_auth(username, password):
return username == ADMIN_USERNAME and password == ADMIN_PASSWORD
def authenticate():
return Response("Login Required", 401, {"WWW-Authenticate": "Basic realm=\"Login Required\""})
def requires_auth(f):
@wraps(f)
def decorated(*args, **kwargs):
auth = request.authorization
if not auth or not check_auth(auth.username, auth.password):
return authenticate()
return f(*args, **kwargs)
return decorated
HTML_FORM = '''
WireGuard Config Generator
WireGuard Config Generator
Welcome, your token is valid. Click below to generate your secure config.
How to use this page
'''
HTML_INSTRUCTIONS = '''
How to Use This Page
📜 How to Use This Page: WireGuard Config Generator
- Go to
https://wgsetup.glorytoyah.org/setup?token=YourToken with your one-time token
- Click "Generate Config" to download your WireGuard config
- Import it into the WireGuard app (mobile or desktop)
'''
HTML_ADMIN_GUIDE = '''
Admin Guide
Admin Guide
- 📦 Token Creation: Use /generate-and-redirect/<GroupName> to generate a token.
- 🔐 Access Token Panel: Visit /admin/tokens to manage tokens.
- 🗃 Used Tokens: View /admin/used for audit logs of used/revoked tokens.
- 🧰 Subnet Setup: Add or modify group subnets via /admin/subnets.
- 💾 Backup: Download backups from /admin/backup.
- ♻️ Restore: Unzip a backup and move files into /opt/wgconfig.
Common Issues:
- 🚫 Permission denied? Ensure correct file permissions (e.g., chown www-data).
- ❌ WireGuard fails to reload? Check for duplicate IPs or malformed entries.
FAQs:
- Can a token be reused? No. Tokens are removed after use.
- Can I edit peer configs manually? Yes, but ensure unique IPs and public keys.
'''
@app.route("/admin/guide")
@requires_auth
def admin_guide():
return render_template_string(HTML_ADMIN_GUIDE)
@app.route("/instructions")
def instructions():
return render_template_string(HTML_INSTRUCTIONS)
@app.route("/generate-and-redirect/")
def generate_and_redirect(group):
tokens = load_tokens()
subnets = load_subnets()
if group not in subnets:
return "Invalid group name", 400
base = subnets[group]
used = [data["ip"] for data in tokens.values() if data.get("group") == group and data.get("ip", "").startswith(base)]
next_host = 2 + len(used)
new_ip = f"{base}.{next_host}/24"
while ip_in_use(new_ip):
next_host += 1
new_ip = f"{base}.{next_host}/24"
token = str(uuid.uuid4())
tokens[token] = {"ip": new_ip, "group": group}
save_tokens(tokens)
return redirect(url_for("setup", token=token))
@app.route("/setup", methods=["GET", "POST"])
def setup():
token = request.args.get("token")
tokens = load_tokens()
if token not in tokens:
return abort(403, description="Invalid or missing token.")
assigned_ip = tokens[token]["ip"]
if request.method == "POST":
try:
private_key = subprocess.check_output(["wg", "genkey"]).decode().strip()
public_key = subprocess.check_output(["wg", "pubkey"], input=private_key.encode()).decode().strip()
except Exception as e:
return f"Key generation failed: {e}", 500
config = f"""
[Interface]
PrivateKey = {private_key}
Address = {assigned_ip}
[Peer]
PublicKey = {SERVER_PUBLIC_KEY}
Endpoint = {SERVER_ENDPOINT}
AllowedIPs = 0.0.0.0/0
PersistentKeepalive = 25
"""
temp_dir = tempfile.mkdtemp()
config_path = os.path.join(temp_dir, f"client_{token}.conf")
with open(config_path, "w") as f:
f.write(config)
with open(KEY_LOG_PATH, "a") as log:
log.write(f"[{datetime.now()}] {token} -> {public_key} ({assigned_ip})\n")
peer_entry = f"\n[Peer]\nPublicKey = {public_key}\nAllowedIPs = {assigned_ip.split('/')[0]}/32\n"
with open(PEERS_CONF_PATH, "a") as peerconf:
peerconf.write(peer_entry)
try:
subprocess.run(["sudo", "wg", "addconf", "wg0", PEERS_CONF_PATH], check=True, stderr=subprocess.PIPE)
except subprocess.CalledProcessError as e:
app.logger.error(f"Failed to update live WireGuard config: {e.stderr.decode()}")
with open(USED_LOG_PATH, "a") as used:
used.write(f"[USED] {token} - {tokens[token]}\n")
tokens.pop(token)
save_tokens(tokens)
return send_file(config_path, as_attachment=True)
return render_template_string(HTML_FORM)
@app.route("/admin/tokens")
@requires_auth
def admin_tokens():
tokens = load_tokens()
html = "Active Tokens
| Token | IP | Group | Revoke |
|---|
"
for t, v in tokens.items():
html += f"| {t} | {v['ip']} | {v['group']} | Revoke |
"
html += "
"
html += "View Used | Subnets | Backup | Admin Guide
"
return html
@app.route("/admin/revoke/")
@requires_auth
def revoke_token(token):
tokens = load_tokens()
if token in tokens:
with open(USED_LOG_PATH, "a") as log:
log.write(f"[REVOKED] {token} - {tokens[token]}\n")
tokens.pop(token)
save_tokens(tokens)
return redirect(url_for("admin_tokens"))
@app.route("/admin/tokens/purge", methods=["POST"])
@requires_auth
def purge_tokens():
open(TOKEN_DB_PATH, "w").write("{}")
return redirect(url_for("admin_tokens"))
@app.route("/admin/subnets", methods=["GET", "POST"])
@requires_auth
def manage_subnets():
subnets = load_subnets()
if request.method == "POST":
name = request.form.get("name")
base = request.form.get("base")
if name and base:
subnets[name] = base
save_subnets(subnets)
html = "Subnets
| Group | Subnet |
|---|
"
for k, v in subnets.items():
html += f"| {k} | {v} |
"
html += "
"
html += "Back to Tokens
"
return html
@app.route("/admin/used")
@requires_auth
def view_used():
if not os.path.exists(USED_LOG_PATH):
return "No used tokens logged.
Back
"
with open(USED_LOG_PATH) as f:
return f"Used/Revoked Tokens
{f.read()}Back
"
@app.route("/admin/backup")
@requires_auth
def backup_all():
os.makedirs(BACKUP_DIR, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
zip_path = os.path.join(BACKUP_DIR, f"backup_{timestamp}.zip")
shutil.make_archive(zip_path.replace(".zip", ""), 'zip', "/opt/wgconfig")
return send_file(zip_path, as_attachment=True)
def load_tokens():
if os.path.exists(TOKEN_DB_PATH):
with open(TOKEN_DB_PATH) as f:
return json.load(f)
return {}
def save_tokens(tokens):
with open(TOKEN_DB_PATH, "w") as f:
json.dump(tokens, f, indent=2)
def load_subnets():
if os.path.exists(SUBNET_DB_PATH):
with open(SUBNET_DB_PATH) as f:
return json.load(f)
return {}
def save_subnets(subnets):
with open(SUBNET_DB_PATH, "w") as f:
json.dump(subnets, f, indent=2)
def ip_in_use(ip):
if not os.path.exists(PEERS_CONF_PATH):
return False
with open(PEERS_CONF_PATH) as f:
return ip.split('/')[0] in f.read()
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)