From f367103ceb4108433375b20ca8ff656fafbb51da Mon Sep 17 00:00:00 2001 From: "ashkan@beigi.net" Date: Wed, 1 Jul 2026 06:11:14 +0000 Subject: [PATCH] Harden: capability-token profile URLs, fast core failure, config validation Security: - Serve generated profiles at /api/profile/ instead of /api/profile/. Profiles carry Ki/OPc, so the enumerable id scheme allowed anyone on the WiFi to harvest all subscriber keys. Unknown tokens return a plain 404 so they can't be probed. Robustness: - Open5GS/free5gc adapters use short MongoDB timeouts so an unreachable core fails provisioning fast instead of blocking ~30s on the default. - Profile cache is now a size-capped LRU (OrderedDict) to prevent unbounded memory growth. - load_config rejects empty/non-mapping YAML and validates network.op_key is a 16-byte hex string. Tests: +25 (test_app.py API coverage, test_utils.py); 78 pass. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 17 ++++++ server/app.py | 39 +++++++++----- server/core_adapters/open5gs.py | 9 +++- server/utils.py | 18 +++++++ tests/test_app.py | 92 +++++++++++++++++++++++++++++++++ tests/test_utils.py | 85 ++++++++++++++++++++++++++++++ 6 files changed, 245 insertions(+), 15 deletions(-) create mode 100644 tests/test_app.py create mode 100644 tests/test_utils.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ca8d616..d5d13fb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Security +- Profile download URLs now use an unguessable capability token + (`secrets.token_urlsafe`) instead of the sequential subscriber id. Profiles + contain Ki/OPc, so the old `/api/profile/` scheme let anyone on the WiFi + enumerate and harvest every subscriber's keys. Unknown tokens return a plain + 404 (indistinguishable from evicted) so tokens cannot be probed. + +### Hardened +- Core adapters set short MongoDB timeouts (`serverSelectionTimeoutMS` etc.) so + an unreachable core fails provisioning fast instead of hanging ~30s. +- In-memory profile cache is now a size-capped LRU (`_MAX_CACHED_PROFILES`) and + can no longer grow without bound. +- `load_config` rejects an empty/non-mapping YAML file and validates + `network.op_key` is a 16-byte hex string, with clear error messages. +- Added `test_app.py` (API-level: provision, rate-limit, token download, + 404 on unknown/enumerated ids) and `test_utils.py` (MAC/IMSI/op_key/config). + ### Added - **Phase-2 profile generation.** `create_esim_profile` now emits a structured profile package with real 3GPP encodings instead of a loose JSON mock: diff --git a/server/app.py b/server/app.py index 4208269..5074f3e 100644 --- a/server/app.py +++ b/server/app.py @@ -1,6 +1,9 @@ +import io import logging import os +import secrets import time +from collections import OrderedDict from datetime import datetime, timezone from flask import Flask, jsonify, redirect, request, send_file, send_from_directory @@ -11,7 +14,6 @@ from server.database import ( check_rate_limit, export_all, get_connection, - get_subscriber_by_id, init_db, ) from server.profile_generator import ( @@ -35,10 +37,21 @@ logger = logging.getLogger(__name__) CONFIG_PATH = os.environ.get("CONFIG_PATH", "config.yaml") -_profile_cache: dict[int, bytes] = {} +# 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() _start_time = time.time() +def _cache_profile(token: str, data: bytes) -> None: + _profile_cache[token] = data + _profile_cache.move_to_end(token) + while len(_profile_cache) > _MAX_CACHED_PROFILES: + _profile_cache.popitem(last=False) + + def create_app(config_path: str = CONFIG_PATH) -> Flask: cfg = load_config(config_path) @@ -120,7 +133,8 @@ def create_app(config_path: str = CONFIG_PATH) -> Flask: ) sub_id = add_subscriber(conn, imsi, iccid, ki_hex, opc_hex, mac) - _profile_cache[sub_id] = profile_bytes + token = secrets.token_urlsafe(24) + _cache_profile(token, profile_bytes) if adapter: try: @@ -134,7 +148,7 @@ def create_app(config_path: str = CONFIG_PATH) -> Flask: "id": sub_id, "imsi": imsi, "iccid": iccid, - "profile_url": f"/api/profile/{sub_id}", + "profile_url": f"/api/profile/{token}", } ), 200 @@ -144,18 +158,15 @@ def create_app(config_path: str = CONFIG_PATH) -> Flask: finally: conn.close() - @app.route("/api/profile/") - def get_profile(sub_id: int): - profile_bytes = _profile_cache.get(sub_id) + @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 = _profile_cache.get(token) if profile_bytes is None: - conn = get_connection(db_path) - sub = get_subscriber_by_id(conn, sub_id) - conn.close() - if sub is None: - return jsonify({"error": "Profile not found"}), 404 - return jsonify({"error": "Profile no longer cached; re-provision to download"}), 410 + return jsonify({"error": "Profile not found"}), 404 - import io return send_file( io.BytesIO(profile_bytes), mimetype="application/octet-stream", diff --git a/server/core_adapters/open5gs.py b/server/core_adapters/open5gs.py index b282817..647ca1f 100644 --- a/server/core_adapters/open5gs.py +++ b/server/core_adapters/open5gs.py @@ -9,7 +9,14 @@ logger = logging.getLogger(__name__) class Open5GSAdapter(CoreAdapter): def __init__(self, mongodb_uri: str, database_name: str = "open5gs"): - self.client = MongoClient(mongodb_uri) + # Short timeouts so a down/unreachable core fails the provision fast + # instead of blocking the request for pymongo's 30s default. + self.client = MongoClient( + mongodb_uri, + serverSelectionTimeoutMS=3000, + connectTimeoutMS=3000, + socketTimeoutMS=3000, + ) self.db = self.client[database_name] def add_subscriber(self, imsi: str, ki: str, opc: str) -> None: diff --git a/server/utils.py b/server/utils.py index 7c3a457..283b540 100644 --- a/server/utils.py +++ b/server/utils.py @@ -34,15 +34,33 @@ def validate_imsi(imsi: str) -> bool: return bool(imsi) and imsi.isdigit() and len(imsi) == 15 +def validate_op_key(op_key: str) -> bool: + """Return True if op_key is a 16-byte (32 hex digit) hex string.""" + if not isinstance(op_key, str) or len(op_key) != 32: + return False + try: + bytes.fromhex(op_key) + except ValueError: + return False + return True + + def load_config(path: str) -> dict: """Load and minimally validate a YAML config file.""" if not os.path.exists(path): raise FileNotFoundError(f"Config file not found: {path}") with open(path) as f: cfg = yaml.safe_load(f) + if not isinstance(cfg, dict): + raise ValueError(f"Config file is empty or not a mapping: {path}") for key in _REQUIRED_CONFIG_KEYS: if key not in cfg: raise ValueError(f"Missing required config section: '{key}'") + op_key = cfg["network"].get("op_key") + if not validate_op_key(op_key): + raise ValueError( + "network.op_key must be a 16-byte hex string (32 hex digits)" + ) return cfg diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..5d0fba0 --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,92 @@ +import json + +import pytest +import yaml + +from server.app import create_app + + +def _write_config(tmp_path): + cfg = { + "network": {"mcc": "302", "mnc": "720", "plmn": "302720", + "op_key": "63bfa50ee6523365ff14c1f45f88737d"}, + "wifi": {"ssid": "Emergency-5G", "channel": 6, "ip": "192.168.4.1", + "subnet": "192.168.4.0/24", "dhcp_range": "192.168.4.10,192.168.4.250"}, + "server": {"host": "0.0.0.0", "port": 5000, "debug": False}, + "database": {"path": str(tmp_path / "subscribers.db")}, + "rate_limiting": {"max_profiles_per_mac": 1, "window_hours": 1}, + "core": {"type": "none", "mongodb_uri": "mongodb://localhost:27017", + "database_name": "open5gs"}, + "profile": {"carrier_name": "Ad-Hoc Network", "profile_name": "Emergency 5G", + "apn": "internet", "smdp_address": "esim.local"}, + } + path = tmp_path / "config.yaml" + path.write_text(yaml.safe_dump(cfg)) + return str(path) + + +@pytest.fixture +def client(tmp_path): + app = create_app(_write_config(tmp_path)) + app.config.update(TESTING=True) + return app.test_client() + + +class TestRoutes: + def test_index_redirects_to_activate(self, client): + r = client.get("/") + assert r.status_code == 302 + assert "/activate" in r.headers["Location"] + + def test_status_ok(self, client): + data = client.get("/api/status").get_json() + assert data["status"] == "ok" + assert data["core_adapter"] == "none" + + def test_export_empty(self, client): + data = client.get("/api/export").get_json() + assert data["count"] == 0 + assert data["subscribers"] == [] + + +class TestProvision: + def test_provision_success(self, client): + r = client.post("/api/provision", json={"mac_address": "aa:bb:cc:11:22:33"}) + assert r.status_code == 200 + body = r.get_json() + assert len(body["imsi"]) == 15 + assert body["profile_url"].startswith("/api/profile/") + + def test_provision_missing_mac(self, client): + # No body and (in tests) no resolvable ARP entry -> 400 + r = client.post("/api/provision", json={}) + assert r.status_code == 400 + + def test_provision_invalid_mac(self, client): + r = client.post("/api/provision", json={"mac_address": "not-a-mac"}) + assert r.status_code == 400 + + def test_rate_limited_second_request(self, client): + client.post("/api/provision", json={"mac_address": "de:ad:be:ef:00:01"}) + r = client.post("/api/provision", json={"mac_address": "de:ad:be:ef:00:01"}) + assert r.status_code == 429 + + +class TestProfileDownload: + def test_download_with_token(self, client): + r = client.post("/api/provision", json={"mac_address": "aa:bb:cc:44:55:66"}) + url = r.get_json()["profile_url"] + d = client.get(url) + assert d.status_code == 200 + profile = json.loads(d.data) + assert profile["type"] == "profile-package-v2" + + def test_unknown_token_is_404(self, client): + assert client.get("/api/profile/does-not-exist").status_code == 404 + + def test_sequential_id_not_downloadable(self, client): + # The old enumerable /api/profile/ scheme must be gone: the row id + # is not a valid download token. + r = client.post("/api/provision", json={"mac_address": "aa:bb:cc:77:88:99"}) + sub_id = r.get_json()["id"] + assert client.get(f"/api/profile/{sub_id}").status_code == 404 diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..2b2920b --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,85 @@ +import pytest + +from server.utils import ( + load_config, + normalize_mac, + validate_imsi, + validate_mac, + validate_op_key, +) + + +class TestMac: + def test_valid(self): + assert validate_mac("aa:bb:cc:dd:ee:ff") + + def test_invalid(self): + assert not validate_mac("nope") + assert not validate_mac("") + + def test_normalize_dashes_and_case(self): + assert normalize_mac("AA-BB-CC-DD-EE-FF") == "aa:bb:cc:dd:ee:ff" + + def test_normalize_no_separator(self): + assert normalize_mac("aabbccddeeff") == "aa:bb:cc:dd:ee:ff" + + +class TestImsi: + def test_valid(self): + assert validate_imsi("302720000000001") + + def test_wrong_length(self): + assert not validate_imsi("30272000000000") + + def test_non_digit(self): + assert not validate_imsi("30272000000000x") + + +class TestOpKey: + def test_valid(self): + assert validate_op_key("63bfa50ee6523365ff14c1f45f88737d") + + def test_wrong_length(self): + assert not validate_op_key("abcd") + + def test_non_hex(self): + assert not validate_op_key("zz" * 16) + + +class TestLoadConfig: + def _base(self): + return ( + "network: {mcc: '302', mnc: '720', op_key: '63bfa50ee6523365ff14c1f45f88737d'}\n" + "server: {host: '0.0.0.0', port: 5000}\n" + "database: {path: '/tmp/x.db'}\n" + "rate_limiting: {window_hours: 1}\n" + "profile: {profile_name: 'P', carrier_name: 'C'}\n" + ) + + def test_valid(self, tmp_path): + p = tmp_path / "c.yaml" + p.write_text(self._base()) + cfg = load_config(str(p)) + assert cfg["network"]["mcc"] == "302" + + def test_missing_file(self): + with pytest.raises(FileNotFoundError): + load_config("/no/such/config.yaml") + + def test_empty_file(self, tmp_path): + p = tmp_path / "empty.yaml" + p.write_text("") + with pytest.raises(ValueError, match="empty or not a mapping"): + load_config(str(p)) + + def test_missing_section(self, tmp_path): + p = tmp_path / "c.yaml" + p.write_text("network: {op_key: '63bfa50ee6523365ff14c1f45f88737d'}\n") + with pytest.raises(ValueError, match="Missing required config section"): + load_config(str(p)) + + def test_bad_op_key(self, tmp_path): + p = tmp_path / "c.yaml" + p.write_text(self._base().replace("63bfa50ee6523365ff14c1f45f88737d", "abcd")) + with pytest.raises(ValueError, match="op_key"): + load_config(str(p))