Harden: fix bare-filename db path crash, configurable APN

- Fix FileNotFoundError when database.path has no directory component
  (os.makedirs(os.path.dirname(path)) with path="subscribers.db" raised).
  Add ensure_parent_dir helper; use it in app.py and scripts/init_db.py.
- Open5GS/free5gc adapters now use profile.apn for the core session/DNN name
  instead of a hardcoded "internet"; apn is threaded through get_adapter.

Tests: +4 (ensure_parent_dir, adapter APN); 82 pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

CHANGELOG.md

@@ -21,8 +21,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   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.
+- Fixed `FileNotFoundError` when `database.path` is a bare filename with no
+  directory component: `app.py` and `scripts/init_db.py` now use a shared
+  `ensure_parent_dir` helper that no-ops on an empty parent.
+- Open5GS/free5gc adapters now use the configured `profile.apn` for the core
+  session/DNN name instead of a hardcoded `"internet"`.
 - 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).
+  404 on unknown/enumerated ids), `test_utils.py` (MAC/IMSI/op_key/config/
+  ensure_parent_dir), and adapter APN coverage.
 
 ### Added
 - **Phase-2 profile generation.** `create_esim_profile` now emits a structured

scripts/init_db.py

@@ -6,14 +6,14 @@ import sys
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 from server.database import init_db
-from server.utils import load_config
+from server.utils import ensure_parent_dir, load_config
 
 
 def main():
     config_path = os.environ.get("CONFIG_PATH", "config.yaml")
     cfg = load_config(config_path)
     db_path = cfg["database"]["path"]
-    os.makedirs(os.path.dirname(db_path), exist_ok=True)
+    ensure_parent_dir(db_path)
     init_db(db_path)
     print(f"Database initialized at {db_path}")
 

server/app.py

@@ -25,6 +25,7 @@ from server.profile_generator import (
 )
 from server.utils import (
     bytes_to_hex,
+    ensure_parent_dir,
     get_client_ip,
     get_mac_from_arp,
     hex_to_bytes,
@@ -61,7 +62,7 @@ def create_app(config_path: str = CONFIG_PATH) -> Flask:
     app = Flask(__name__, static_folder=portal_dir, static_url_path="")
 
     db_path = cfg["database"]["path"]
-    os.makedirs(os.path.dirname(db_path), exist_ok=True)
+    ensure_parent_dir(db_path)
     init_db(db_path)
 
     adapter = get_adapter(cfg)

server/core_adapters/__init__.py

@@ -8,12 +8,13 @@ def get_adapter(config: dict) -> Optional[CoreAdapter]:
     core_type = core_cfg.get("type", "none")
     uri = core_cfg.get("mongodb_uri", "mongodb://localhost:27017")
     db_name = core_cfg.get("database_name", "open5gs")
+    apn = config.get("profile", {}).get("apn", "internet")
 
     if core_type == "open5gs":
         from .open5gs import Open5GSAdapter
-        return Open5GSAdapter(uri, db_name)
+        return Open5GSAdapter(uri, db_name, apn)
     elif core_type == "free5gc":
         from .free5gc import Free5GCAdapter
-        return Free5GCAdapter(uri, db_name)
+        return Free5GCAdapter(uri, db_name, apn)
     else:
         return None

server/core_adapters/free5gc.py

@@ -7,5 +7,6 @@ class Free5GCAdapter(Open5GSAdapter):
     Override methods here as free5gc-specific behaviour is needed.
     """
 
-    def __init__(self, mongodb_uri: str, database_name: str = "free5gc"):
+    def __init__(self, mongodb_uri: str, database_name: str = "free5gc",
-        super().__init__(mongodb_uri, database_name)
+                 apn: str = "internet"):
+        super().__init__(mongodb_uri, database_name, apn)

server/core_adapters/open5gs.py

@@ -8,7 +8,8 @@ logger = logging.getLogger(__name__)
 
 
 class Open5GSAdapter(CoreAdapter):
-    def __init__(self, mongodb_uri: str, database_name: str = "open5gs"):
+    def __init__(self, mongodb_uri: str, database_name: str = "open5gs",
+                 apn: str = "internet"):
         # Short timeouts so a down/unreachable core fails the provision fast
         # instead of blocking the request for pymongo's 30s default.
         self.client = MongoClient(
@@ -18,6 +19,7 @@ class Open5GSAdapter(CoreAdapter):
             socketTimeoutMS=3000,
         )
         self.db = self.client[database_name]
+        self.apn = apn
 
     def add_subscriber(self, imsi: str, ki: str, opc: str) -> None:
         """Upsert a subscriber into the Open5GS MongoDB collection."""
@@ -39,7 +41,7 @@ class Open5GSAdapter(CoreAdapter):
                     "default_indicator": True,
                     "session": [
                         {
-                            "name": "internet",
+                            "name": self.apn,
                             "type": 3,
                             "ambr": {
                                 "downlink": {"value": 1, "unit": 3},

server/utils.py

@@ -64,6 +64,17 @@ def load_config(path: str) -> dict:
     return cfg
 
 
+def ensure_parent_dir(path: str) -> None:
+    """Create the parent directory of ``path`` if it has one.
+
+    Guards against ``os.makedirs("")`` raising FileNotFoundError when ``path``
+    is a bare filename with no directory component.
+    """
+    parent = os.path.dirname(path)
+    if parent:
+        os.makedirs(parent, exist_ok=True)
+
+
 def hex_to_bytes(h: str) -> bytes:
     return bytes.fromhex(h)
 

tests/test_core_adapters.py

@@ -26,6 +26,12 @@ class TestFactory:
             adapter = get_adapter(cfg)
         assert isinstance(adapter, Free5GCAdapter)
 
+    def test_apn_threaded_from_profile_config(self):
+        cfg = {"core": {"type": "open5gs"}, "profile": {"apn": "ims"}}
+        with patch("server.core_adapters.open5gs.MongoClient"):
+            adapter = get_adapter(cfg)
+        assert adapter.apn == "ims"
+
 
 @pytest.fixture
 def open5gs_adapter():
@@ -56,6 +62,14 @@ class TestOpen5GSAdapter:
         assert sub_doc["slice"][0]["sst"] == 1
         assert sub_doc["slice"][0]["session"][0]["name"] == "internet"
 
+    def test_add_subscriber_uses_configured_apn(self):
+        with patch("server.core_adapters.open5gs.MongoClient"):
+            adapter = Open5GSAdapter("mongodb://localhost:27017", "open5gs", apn="ims")
+            adapter.db = MagicMock()
+        adapter.add_subscriber("302720000000001", "aa", "bb")
+        sub_doc = adapter.db.subscribers.replace_one.call_args[0][1]
+        assert sub_doc["slice"][0]["session"][0]["name"] == "ims"
+
     def test_add_subscriber_upsert_true(self, open5gs_adapter):
         open5gs_adapter.add_subscriber("302720000000001", "aa", "bb")
         call_kwargs = open5gs_adapter.db.subscribers.replace_one.call_args[1]

tests/test_utils.py

@@ -1,6 +1,7 @@
 import pytest
 
 from server.utils import (
+    ensure_parent_dir,
     load_config,
     normalize_mac,
     validate_imsi,
@@ -9,6 +10,17 @@ from server.utils import (
 )
 
 
+class TestEnsureParentDir:
+    def test_bare_filename_does_not_crash(self):
+        # No directory component -> must be a no-op, not FileNotFoundError.
+        ensure_parent_dir("subscribers.db")
+
+    def test_creates_nested_dirs(self, tmp_path):
+        target = tmp_path / "a" / "b" / "x.db"
+        ensure_parent_dir(str(target))
+        assert (tmp_path / "a" / "b").is_dir()
+
+
 class TestMac:
     def test_valid(self):
         assert validate_mac("aa:bb:cc:dd:ee:ff")