Push Tracker
ria-ran--sim-drop/plan/05-flask-app.md
A ashkan@beigi.net 8672b0de6b Add Phase-1 eSIM provisioning implementation
Full working closed-loop eSIM provisioning stack (spec in claude.md / plan/):
Flask API, SQLite subscriber store, credential/profile generator, Open5GS
and free5gc core adapters, captive portal, operational scripts, systemd
units, network configs, and a 44-test suite.

Profile generation currently returns a mock JSON profile; real SGP.22
generation via pySim is the planned Phase-2 upgrade.

Fixes two bugs found while getting the suite green:
- generate_imsi hardcoded a 10-digit MSIN, producing 16-digit IMSIs for a
  3-digit MNC; MSIN length is now 15 - len(mcc) - len(mnc).
- check_rate_limit compared timestamps as strings across mismatched formats
  (SQLite CURRENT_TIMESTAMP vs Python isoformat); both sides now normalized
  via SQLite datetime() so the window check is chronologically correct.

Add .gitignore for venv, caches, runtime config.yaml, and databases.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-01 05:56:31 +00:00

4.2 KiB

Unit 05 — Flask Application

Goal

Main Flask server wiring together the database, profile generator, utils, and core adapters.

File

  • server/app.py

Endpoints

Method Path Auth Description
GET / none 302 redirect → /activate
GET /activate none Serve portal/index.html
POST /api/provision none Generate + provision eSIM
GET /api/profile/<int:id> none Download profile bytes
GET /api/export none JSON export of all subscribers
GET /api/status none Health check

POST /api/provision — Full Flow

Request body (JSON):

{"mac_address": "aa:bb:cc:dd:ee:ff"}

Steps:

  1. Parse JSON body; if mac_address missing/empty, resolve from server-side ARP (/proc/net/arp lookup by client IP)
  2. normalize_mac() + validate_mac() → 400 if invalid
  3. check_rate_limit(conn, mac, window_hours) → 429 if exceeded
  4. generate_imsi(mcc, mnc)
  5. generate_ki()
  6. generate_iccid()
  7. derive_opc(ki, op_bytes)
  8. create_esim_profile(...)profile_bytes
  9. add_subscriber(conn, imsi, iccid, ki_hex, opc_hex, mac)sub_id
  10. Store profile_bytes in memory cache {sub_id: bytes} (or temp file)
  11. If core_adapter configured: attempt add_subscriber(imsi, ki_hex, opc_hex); log failure, don't abort
  12. Return 200:
{
  "id": 1,
  "imsi": "302720XXXXXXXXXX",
  "iccid": "8910302720XXXXXXXXX",
  "profile_url": "/api/profile/1"
}

Error responses:

  • 400: {"error": "Invalid MAC address"}
  • 429: {"error": "Rate limit exceeded. One profile per device per hour."}
  • 500: {"error": "Profile generation failed"}

GET /api/profile/<id> — Profile Download

  • Look up subscriber by id in DB
  • Retrieve profile bytes from in-memory cache (keyed by sub_id)
  • 404 if not found or cache expired
  • Response: Content-Type: application/octet-stream, Content-Disposition: attachment; filename=esim-profile.bin

GET /api/export — Subscriber Export

Response:

{
  "exported_at": "2025-02-11T10:30:00Z",
  "count": 123,
  "subscribers": [
    {
      "imsi": "302720123456789",
      "ki": "00112233445566778899aabbccddeeff",
      "opc": "63bfa50ee6523365ff14c1f45f88737d",
      "iccid": "8910302720123456789",
      "mac_address": "aa:bb:cc:dd:ee:ff",
      "created_at": "2025-02-11T09:15:00Z",
      "synced_to_core": false
    }
  ]
}

GET /api/status — Health

{
  "status": "ok",
  "subscriber_count": 42,
  "uptime_seconds": 3600,
  "core_adapter": "open5gs"
}

Application Setup

import os, logging
from flask import Flask, request, jsonify, send_file, redirect
from server.database import init_db, get_connection, ...
from server.profile_generator import ...
from server.utils import load_config, ...
from server.core_adapters import get_adapter

CONFIG_PATH = os.environ.get("CONFIG_PATH", "config.yaml")

def create_app(config_path=CONFIG_PATH):
    app = Flask(__name__, static_folder="../portal", static_url_path="")
    cfg = load_config(config_path)

    db_path = cfg["database"]["path"]
    init_db(db_path)

    adapter = get_adapter(cfg)
    profile_cache = {}   # {sub_id: bytes}
    start_time = time.time()

    # register routes ...
    return app

if __name__ == "__main__":
    app = create_app()
    app.run(host=cfg["server"]["host"], port=cfg["server"]["port"])

MAC Address from ARP (server-side fallback)

When client doesn't send MAC:

def get_mac_from_arp(ip: str) -> str | None:
    try:
        with open("/proc/net/arp") as f:
            for line in f:
                parts = line.split()
                if len(parts) >= 4 and parts[0] == ip:
                    return parts[3]
    except Exception:
        pass
    return None

Verification

CONFIG_PATH=config.yaml python -m server.app &
sleep 1
curl -s http://localhost:5000/api/status | python -m json.tool
curl -s -X POST http://localhost:5000/api/provision \
     -H "Content-Type: application/json" \
     -d '{"mac_address":"aa:bb:cc:dd:ee:ff"}' | python -m json.tool
curl -s http://localhost:5000/api/export | python -m json.tool
kill %1