Harden: gate /api/export behind an admin token (closed by default)
/api/export returns every subscriber's Ki/OPc and was reachable unauthenticated by any WiFi client (and directly, since Flask binds 0.0.0.0). It now requires server.admin_token via the X-Admin-Token header (constant-time compare) and is disabled entirely when no token is configured. Add admin_token to config.yaml.example. Tests: +2 (disabled/401/200); 86 pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
parent
035bc59460
commit
0b362bbd0b
|
|
@ -8,6 +8,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||||
## [Unreleased]
|
## [Unreleased]
|
||||||
|
|
||||||
### Security
|
### Security
|
||||||
|
- `/api/export` (dumps every subscriber's Ki/OPc) now requires an admin token
|
||||||
|
(`server.admin_token`, sent as `X-Admin-Token`, constant-time compared) and
|
||||||
|
is disabled by default. Previously any WiFi client could pull all credentials
|
||||||
|
unauthenticated.
|
||||||
- Profile download URLs now use an unguessable capability token
|
- Profile download URLs now use an unguessable capability token
|
||||||
(`secrets.token_urlsafe`) instead of the sequential subscriber id. Profiles
|
(`secrets.token_urlsafe`) instead of the sequential subscriber id. Profiles
|
||||||
contain Ki/OPc, so the old `/api/profile/<int>` scheme let anyone on the WiFi
|
contain Ki/OPc, so the old `/api/profile/<int>` scheme let anyone on the WiFi
|
||||||
|
|
|
||||||
|
|
@ -15,6 +15,7 @@ server:
|
||||||
host: "0.0.0.0"
|
host: "0.0.0.0"
|
||||||
port: 5000
|
port: 5000
|
||||||
debug: false
|
debug: false
|
||||||
|
admin_token: "" # set a secret to enable /api/export (sent as X-Admin-Token); empty = export disabled
|
||||||
|
|
||||||
database:
|
database:
|
||||||
path: "/opt/esim-provisioning/data/subscribers.db"
|
path: "/opt/esim-provisioning/data/subscribers.db"
|
||||||
|
|
|
||||||
|
|
@ -93,6 +93,9 @@ def create_app(config_path: str = CONFIG_PATH) -> Flask:
|
||||||
rl_cfg = cfg["rate_limiting"]
|
rl_cfg = cfg["rate_limiting"]
|
||||||
window_hours = rl_cfg["window_hours"]
|
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
|
# Routes
|
||||||
# ------------------------------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
|
|
@ -194,6 +197,14 @@ def create_app(config_path: str = CONFIG_PATH) -> Flask:
|
||||||
|
|
||||||
@app.route("/api/export")
|
@app.route("/api/export")
|
||||||
def 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)
|
conn = get_connection(db_path)
|
||||||
try:
|
try:
|
||||||
subscribers = export_all(conn)
|
subscribers = export_all(conn)
|
||||||
|
|
|
||||||
|
|
@ -6,13 +6,14 @@ import yaml
|
||||||
from server.app import create_app
|
from server.app import create_app
|
||||||
|
|
||||||
|
|
||||||
def _write_config(tmp_path):
|
def _write_config(tmp_path, admin_token=""):
|
||||||
cfg = {
|
cfg = {
|
||||||
"network": {"mcc": "302", "mnc": "720", "plmn": "302720",
|
"network": {"mcc": "302", "mnc": "720", "plmn": "302720",
|
||||||
"op_key": "63bfa50ee6523365ff14c1f45f88737d"},
|
"op_key": "63bfa50ee6523365ff14c1f45f88737d"},
|
||||||
"wifi": {"ssid": "Emergency-5G", "channel": 6, "ip": "192.168.4.1",
|
"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"},
|
"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},
|
"server": {"host": "0.0.0.0", "port": 5000, "debug": False,
|
||||||
|
"admin_token": admin_token},
|
||||||
"database": {"path": str(tmp_path / "subscribers.db")},
|
"database": {"path": str(tmp_path / "subscribers.db")},
|
||||||
"rate_limiting": {"max_profiles_per_mac": 1, "window_hours": 1},
|
"rate_limiting": {"max_profiles_per_mac": 1, "window_hours": 1},
|
||||||
"core": {"type": "none", "mongodb_uri": "mongodb://localhost:27017",
|
"core": {"type": "none", "mongodb_uri": "mongodb://localhost:27017",
|
||||||
|
|
@ -32,6 +33,13 @@ def client(tmp_path):
|
||||||
return app.test_client()
|
return app.test_client()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def admin_client(tmp_path):
|
||||||
|
app = create_app(_write_config(tmp_path, admin_token="s3cret-token"))
|
||||||
|
app.config.update(TESTING=True)
|
||||||
|
return app.test_client()
|
||||||
|
|
||||||
|
|
||||||
class TestRoutes:
|
class TestRoutes:
|
||||||
def test_index_redirects_to_activate(self, client):
|
def test_index_redirects_to_activate(self, client):
|
||||||
r = client.get("/")
|
r = client.get("/")
|
||||||
|
|
@ -43,8 +51,20 @@ class TestRoutes:
|
||||||
assert data["status"] == "ok"
|
assert data["status"] == "ok"
|
||||||
assert data["core_adapter"] == "none"
|
assert data["core_adapter"] == "none"
|
||||||
|
|
||||||
def test_export_empty(self, client):
|
def test_export_disabled_without_admin_token(self, client):
|
||||||
data = client.get("/api/export").get_json()
|
# No server.admin_token configured -> export is disabled.
|
||||||
|
assert client.get("/api/export").status_code == 403
|
||||||
|
|
||||||
|
def test_export_requires_matching_token(self, admin_client):
|
||||||
|
assert admin_client.get("/api/export").status_code == 401
|
||||||
|
assert admin_client.get(
|
||||||
|
"/api/export", headers={"X-Admin-Token": "wrong"}
|
||||||
|
).status_code == 401
|
||||||
|
|
||||||
|
def test_export_with_valid_token(self, admin_client):
|
||||||
|
r = admin_client.get("/api/export", headers={"X-Admin-Token": "s3cret-token"})
|
||||||
|
assert r.status_code == 200
|
||||||
|
data = r.get_json()
|
||||||
assert data["count"] == 0
|
assert data["count"] == 0
|
||||||
assert data["subscribers"] == []
|
assert data["subscribers"] == []
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue
Block a user