import io import logging import os import secrets import threading import time from collections import OrderedDict from datetime import datetime, timezone from flask import Flask, jsonify, redirect, request, send_file, send_from_directory from server.core_adapters import get_adapter from server.database import ( add_subscriber, check_rate_limit, export_all, get_connection, init_db, ) from server.profile_generator import ( create_esim_profile, derive_opc, generate_iccid, generate_imsi, generate_ki, ) from server.saip_profile import create_saip_profile, saip_available from server.utils import ( bytes_to_hex, ensure_parent_dir, get_client_ip, get_mac_from_arp, hex_to_bytes, load_config, normalize_mac, validate_mac, ) logger = logging.getLogger(__name__) CONFIG_PATH = os.environ.get("CONFIG_PATH", "config.yaml") # Generated profiles are held in memory, keyed by an unguessable download token # (never by the sequential subscriber id — the profile body contains Ki/OPc). # The cache is size-capped so it cannot grow without bound. _MAX_CACHED_PROFILES = 256 _profile_cache: "OrderedDict[str, bytes]" = OrderedDict() _cache_lock = threading.Lock() _start_time = time.time() def _cache_profile(token: str, data: bytes) -> None: with _cache_lock: _profile_cache[token] = data _profile_cache.move_to_end(token) while len(_profile_cache) > _MAX_CACHED_PROFILES: _profile_cache.popitem(last=False) def _get_cached_profile(token: str): with _cache_lock: return _profile_cache.get(token) def create_app(config_path: str = CONFIG_PATH) -> Flask: cfg = load_config(config_path) portal_dir = os.path.join(os.path.dirname(__file__), "..", "portal") portal_dir = os.path.abspath(portal_dir) app = Flask(__name__, static_folder=portal_dir, static_url_path="") db_path = cfg["database"]["path"] ensure_parent_dir(db_path) init_db(db_path) adapter = get_adapter(cfg) if adapter: logger.info(f"Core adapter: {adapter.__class__.__name__}") else: logger.info("Core adapter: none (export-only mode)") net_cfg = cfg["network"] mcc = net_cfg["mcc"] mnc = net_cfg["mnc"] op_bytes = hex_to_bytes(net_cfg["op_key"]) profile_cfg = cfg["profile"] profile_name = profile_cfg["profile_name"] carrier_name = profile_cfg["carrier_name"] apn = profile_cfg.get("apn", "internet") smdp_address = profile_cfg.get("smdp_address", "esim.local") # Profile package format: "v2" (JSON, always available) or "saip" (real # SGP.22 DER). Fall back to v2 if saip is requested but unavailable. profile_format = profile_cfg.get("format", "v2") use_saip = profile_format == "saip" and saip_available() if profile_format == "saip" and not use_saip: logger.warning("profile.format=saip requested but SAIP is unavailable; " "falling back to profile-package-v2") elif use_saip: logger.info("Profile format: SAIP (real SGP.22 DER)") rl_cfg = cfg["rate_limiting"] window_hours = rl_cfg["window_hours"] # Optional admin token guarding /api/export; unset means export is disabled. admin_token = cfg.get("server", {}).get("admin_token") or "" # ------------------------------------------------------------------ # Routes # ------------------------------------------------------------------ @app.route("/") def index(): return redirect("/activate", code=302) @app.route("/activate") def activate(): return send_from_directory(portal_dir, "index.html") @app.route("/api/provision", methods=["POST"]) def provision(): body = request.get_json(silent=True) or {} mac = body.get("mac_address", "").strip() if not mac: client_ip = get_client_ip(request) mac = get_mac_from_arp(client_ip) or "" if not mac: return jsonify({"error": "Could not determine MAC address"}), 400 mac = normalize_mac(mac) if not validate_mac(mac): return jsonify({"error": "Invalid MAC address"}), 400 conn = get_connection(db_path) try: if not check_rate_limit(conn, mac, window_hours): return jsonify( {"error": "Rate limit exceeded. One eSIM profile per device per hour."} ), 429 # Retry on the astronomically-unlikely IMSI/ICCID collision so a # duplicate does not surface to the user as a 500. sub_id = None for attempt in range(5): imsi = generate_imsi(mcc, mnc) ki = generate_ki() iccid = generate_iccid() opc = derive_opc(ki, op_bytes) ki_hex = bytes_to_hex(ki) opc_hex = bytes_to_hex(opc) try: sub_id = add_subscriber(conn, imsi, iccid, ki_hex, opc_hex, mac) break except ValueError as exc: logger.warning(f"Credential collision (attempt {attempt + 1}), retrying: {exc}") if sub_id is None: logger.error("Provision failed: could not allocate a unique IMSI/ICCID") return jsonify({"error": "Profile generation failed"}), 500 if use_saip: profile_bytes = create_saip_profile( imsi, ki, opc, iccid, profile_name, carrier_name, ) else: profile_bytes = create_esim_profile( imsi, ki, opc, iccid, profile_name, carrier_name, smdp_address=smdp_address, apn=apn, ) token = secrets.token_urlsafe(24) _cache_profile(token, profile_bytes) if adapter: try: adapter.add_subscriber(imsi, ki_hex, opc_hex) except Exception as exc: logger.error(f"Core adapter sync failed for {imsi}: {exc}") logger.info(f"Provisioned IMSI {imsi} for MAC {mac} (id={sub_id})") return jsonify( { "id": sub_id, "imsi": imsi, "iccid": iccid, "profile_url": f"/api/profile/{token}", } ), 200 except Exception as exc: logger.exception(f"Provision failed: {exc}") return jsonify({"error": "Profile generation failed"}), 500 finally: conn.close() @app.route("/api/profile/") def get_profile(token: str): # Capability URL: only the holder of the random token can download the # profile. Never distinguish "wrong token" from "expired/evicted" — # both are a plain 404 so tokens can't be probed. profile_bytes = _get_cached_profile(token) if profile_bytes is None: return jsonify({"error": "Profile not found"}), 404 return send_file( io.BytesIO(profile_bytes), mimetype="application/octet-stream", as_attachment=True, download_name="esim-profile.bin", ) @app.route("/api/export") def export(): # This endpoint dumps every subscriber's Ki/OPc, so it is gated behind # an admin token and defaults closed (disabled unless configured). if not admin_token: return jsonify({"error": "Export disabled (server.admin_token not set)"}), 403 provided = request.headers.get("X-Admin-Token", "") if not secrets.compare_digest(provided, admin_token): return jsonify({"error": "Unauthorized"}), 401 conn = get_connection(db_path) try: subscribers = export_all(conn) finally: conn.close() return jsonify( { "exported_at": datetime.now(timezone.utc).isoformat(), "count": len(subscribers), "subscribers": subscribers, } ) @app.route("/api/status") def status(): conn = get_connection(db_path) try: count = conn.execute("SELECT COUNT(*) FROM subscribers").fetchone()[0] finally: conn.close() return jsonify( { "status": "ok", "subscriber_count": count, "uptime_seconds": int(time.time() - _start_time), "core_adapter": adapter.__class__.__name__ if adapter else "none", } ) return app if __name__ == "__main__": logging.basicConfig(level=logging.INFO) cfg = load_config(CONFIG_PATH) app = create_app(CONFIG_PATH) app.run( host=cfg["server"]["host"], port=cfg["server"]["port"], debug=cfg["server"].get("debug", False), )