LarryLaffer-dev · LarryLaffer-dev · May 11, 2026
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -9,7 +9,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Per-interface APN allowlist on `subscriber`: new `apn_list_swx` column (comma-separated APN IDs) controls which APNs are returned over SWx, separate from the existing `apn_list` used on S6a. Exposed through the REST API (auto-generated `SUBSCRIBER` schema), a new "Allowed APNs (SWx / untrusted ePDG access)" multi-select on the HSS GUI subscriber form, and a `databaseSchema` v3 upgrade that ALTERs existing databases (PyHSS `1.0.3`).
 - Documentation: mid-session Gx RAR / `PUT /pcrf/` PCC rule install in `docs/PCRF_Notes.md`
+
+### Changed
+
+- SWx Server-Assignment-Answer (`Answer_16777265_301`) now expands the subscriber's `apn_list_swx` into one `APN-Configuration` AVP per allowed APN inside `Non-3GPP-User-Data`, instead of always returning a single hardcoded `ims` APN. The top-level Non-3GPP-User-Data AMBR now reflects the subscriber's UE-AMBR (was hardcoded 50/100 Mbit/s).
+
+### Breaking
+
+- SWx Server-Assignment-Request is now rejected with Experimental-Result `DIAMETER_ERROR_USER_NO_NON_3GPP_SUBSCRIPTION (5450)` (3GPP TS 29.273 §5.2.2.4) for any subscriber whose `apn_list_swx` is NULL or empty. After upgrading, operators must populate `apn_list_swx` (typically with the IMS APN's id) on every subscriber that should be allowed VoWiFi/ePDG attach -- running the schema migration alone is not enough.
 - 2G / 3G support via Osmocom GSUP.
 - Support for running PyHSS services in Docker containers and provide official Docker images.
 - Database types postgresql and sqlite.

diff --git a/docs/attributes.md b/docs/attributes.md
@@ -38,6 +38,7 @@ We can also get this information in the `/subscriber/imsi/` and `/subscriber/msi
   "imsi": "0010100000000101",
   "auc_id": 1,
   "apn_list": "1,2,3",
+  "apn_list_swx": "1",
   "ue_ambr_dl": 9999999,
   "nam": 0,
   "serving_mme": null,

diff --git a/docs/provisioning.md b/docs/provisioning.md
@@ -39,6 +39,12 @@ curl -X 'PUT' \
 
 ### Define Subscriber for EPC Access
 This defines IMSI 001010000000001 with access to APN with APN_ID 1 & APN_ID 2, where the APN with APN_ID 1 is the default APN for the subscriber. The AMBR values allow for 9999999 bytes per second (~8Mbps).
+
+The subscriber's allowed APNs are configured per Diameter interface:
+
+- `apn_list` -- comma-separated APN IDs returned in the S6a Update-Location-Answer (mobile / 3GPP access via MME). This field is required.
+- `apn_list_swx` -- comma-separated APN IDs returned in the SWx Server-Assignment-Answer (untrusted non-3GPP access via ePDG). Optional. When NULL or empty the SWx SAA is rejected with `DIAMETER_ERROR_USER_NO_NON_3GPP_SUBSCRIPTION (5450)` per 3GPP TS 29.273 §5.2.2.4, so VoWiFi attach is denied. Operators that want VoWiFi must list at least the IMS APN here.
+
 ```shell
 curl -X 'PUT' \
   'http://10.97.0.36:8080/subscriber/' \
@@ -49,7 +55,8 @@ curl -X 'PUT' \
   "enabled": true,
   "auc_id": 1,
   "default_apn": 1,
-  "apn_list": "1",
+  "apn_list": "1,2",
+  "apn_list_swx": "1",
   "msisdn": "599416501",
   "ue_ambr_dl": 9999999,
   "ue_ambr_ul": 9999999,

diff --git a/lib/database.py b/lib/database.py
@@ -117,7 +117,8 @@ class SUBSCRIBER(Base):
     enabled = Column(Boolean, default=1, doc='Subscriber enabled/disabled')
     auc_id = Column(Integer, ForeignKey('auc.auc_id'), doc='Reference to AuC ID defined with SIM Auth data', nullable=False)
     default_apn = Column(Integer, ForeignKey('apn.apn_id'), doc='APN ID to use for the default APN', nullable=False)
-    apn_list = Column(String(64), doc='Comma separated list of allowed APNs', nullable=False)
+    apn_list = Column(String(64), doc='Comma separated list of APN IDs allowed via S6a (mobile access)', nullable=False)
+    apn_list_swx = Column(String(64), doc='Comma separated list of APN IDs allowed via SWx (untrusted access via ePDG); NULL or empty = SWx access denied (DIAMETER_ERROR_USER_NO_NON_3GPP_SUBSCRIPTION)', nullable=True, default=None)
     msisdn = Column(String(18), doc='Primary Phone number of Subscriber')
     ue_ambr_dl = Column(Integer, default=999999, doc='Downlink Aggregate Maximum Bit Rate')
     ue_ambr_ul = Column(Integer, default=999999, doc='Uplink Aggregate Maximum Bit Rate')
@@ -562,8 +563,12 @@ def log_change(self, session, item_id, operation, changes, table_name, operation
         # Combine all changes into a single string with their types
         changes_string = '\r\n\r\n'.join(f"{column_name}: [{type(old_value).__name__}] {old_value} ----> [{type(new_value).__name__}] {new_value}" for column_name, old_value, new_value in changes)
 
+        # Do not use the nasty "or" expression, item_id may be 0
+        if item_id is None:
+            item_id = generated_id
+
         change = OPERATION_LOG_BASE(
-            item_id=item_id or generated_id,
+            item_id=item_id,
             operation_id=operation_id,
             operation=operation,
             last_modified=datetime.datetime.now(tz=timezone.utc),
@@ -604,9 +609,6 @@ def log_changes_before_commit(self, session):
                 if isinstance(obj, OPERATION_LOG_BASE):
                     continue  # Skip change log entries
 
-                item_id = getattr(obj, list(obj.__table__.primary_key.columns.keys())[0])
-                generated_id = None
-
                 #Avoid logging rollback operations
                 if operation == 'ROLLBACK':
                     return
@@ -615,6 +617,10 @@ def log_changes_before_commit(self, session):
                 if operation == 'INSERT':
                     session.flush()
 
+                # Retrieve the attribute after session flush
+                item_id = getattr(obj, list(obj.__table__.primary_key.columns.keys())[0])
+                generated_id = None
+
                 if operation == 'UPDATE':
                     changes = []
                     for attr in class_mapper(obj.__class__).column_attrs:

diff --git a/lib/databaseSchema.py b/lib/databaseSchema.py
@@ -8,7 +8,7 @@
 
 
 class DatabaseSchema:
-    latest = 2
+    latest = 3
 
     def __init__(self, logTool, base, engine: Engine, main_service: bool):
         self.logTool = logTool
@@ -227,6 +227,14 @@ def upgrade_add_ifc_template(self):
         self.add_column("serving_apn", "af_subscriptions", "VARCHAR(1024)")
         self.set_version(2)
 
+    def upgrade_add_apn_list_swx(self):
+        if self.get_version() >= 3:
+            return
+        self.upgrade_msg(3)
+        self.add_column("subscriber", "apn_list_swx", "VARCHAR(64)")
+        self.set_version(3)
+
     def upgrade_all(self):
         self.upgrade_from_20240603_release_1_0_1()
         self.upgrade_add_ifc_template()
+        self.upgrade_add_apn_list_swx()
diff --git a/lib/diameter.py b/lib/diameter.py
@@ -3697,28 +3697,84 @@ def Answer_16777265_301(self, packet_vars, avps):
 
         self.logTool.log(service='HSS', level='debug', message="SWx SAR: IMSI=" + str(imsi) + " assignment_type=" + str(server_assignment_type), redisClient=self.redisMessaging)
 
-        # Build Non-3GPP-User-Data AVP (vendor 10415, AVP 1500)
+        # Per-interface APN allowlist: SWx uses subscriber.apn_list_swx (separate from S6a apn_list).
+        # Per 3GPP TS 29.273 §5.2.2.4, when the subscriber has no non-3GPP subscription
+        # (no APN allowed via SWx), reject with DIAMETER_ERROR_USER_NO_NON_3GPP_SUBSCRIPTION (5450).
+        apn_list_swx_raw = subscriber_details.get('apn_list_swx') or ''
+        swx_apn_ids = [x.strip() for x in str(apn_list_swx_raw).split(',') if x.strip()]
+        if not swx_apn_ids:
+            self.logTool.log(service='HSS', level='info',
+                             message="SWx SAR: IMSI=" + str(imsi) + " has no apn_list_swx; rejecting SAA with DIAMETER_ERROR_USER_NO_NON_3GPP_SUBSCRIPTION (5450)",
+                             redisClient=self.redisMessaging)
+            experimental_result = self.generate_vendor_avp(266, 40, 10415, "")
+            experimental_result += self.generate_avp(298, 40, self.int_to_hex(5450, 4))
+            avp += self.generate_avp(297, 40, experimental_result)
+            self.redisMessaging.sendMetric(serviceName='diameter', metricName='prom_diam_auth_event_count',
+                                            metricType='counter', metricAction='inc', metricValue=1.0,
+                                            metricLabels={"diameter_application_id": 16777265, "diameter_cmd_code": 301, "event": "NoNon3gppSubscription", "imsi_prefix": str(imsi[0:6])},
+                                            metricHelp='Diameter Authentication related Counters',
+                                            metricExpiry=60, usePrefix=True, prefixHostname=self.hostname, prefixServiceName='metric')
+            response = self.generate_diameter_packet("01", "40", 301, 16777265, packet_vars['hop-by-hop-identifier'], packet_vars['end-to-end-identifier'], avp)
+            return response
+
+        # Place default_apn first if it appears in the SWx allowlist, otherwise keep declared order.
+        default_apn = subscriber_details.get('default_apn')
+        if default_apn is not None and str(default_apn) in swx_apn_ids:
+            swx_apn_ids = [str(default_apn)] + [x for x in swx_apn_ids if x != str(default_apn)]
+
+        # Build Non-3GPP-User-Data AVP (vendor 10415, AVP 1500) per TS 29.273 §8.2.3.1
         # Non-3GPP-IP-Access (AVP 1501): NON_3GPP_SUBSCRIPTION_ALLOWED (0)
         non_3gpp_ip_access = self.generate_vendor_avp(1501, "c0", 10415, format(int(0),"x").zfill(8))
         # Non-3GPP-IP-Access-APN (AVP 1502): NON_3GPP_APNS_ENABLE (0)
         non_3gpp_ip_access_apn = self.generate_vendor_avp(1502, "c0", 10415, format(int(0),"x").zfill(8))
         # AN-Trusted (AVP 1503): UNTRUSTED (1) — ePDG access is always untrusted
         an_trusted = self.generate_vendor_avp(1503, "c0", 10415, format(int(1),"x").zfill(8))
 
-        # APN-Configuration (AVP 1430) for 'ims'
-        apn_context_id = self.generate_vendor_avp(1423, "c0", 10415, format(int(0),"x").zfill(8))
-        apn_service_selection = self.generate_avp(493, 40, str(binascii.hexlify(b'ims'),'ascii'))
-        apn_pdn_type = self.generate_vendor_avp(1456, "c0", 10415, format(int(0),"x").zfill(8))  # IPv4
-        apn_config = self.generate_vendor_avp(1430, "c0", 10415, apn_context_id + apn_service_selection + apn_pdn_type)
+        # APN-Configuration-Profile (AVP 1429) wraps one APN-Configuration (AVP 1430) per allowed APN.
+        apn_configurations = ''
+        apn_context_identifer_count = 1
+        for apn_id in swx_apn_ids:
+            try:
+                apn_data = self.database.Get_APN(apn_id)
+            except Exception as e:
+                self.logTool.log(service='HSS', level='error',
+                                 message="SWx SAR: failed to load APN id " + str(apn_id) + ": " + str(e),
+                                 redisClient=self.redisMessaging)
+                continue
 
-        # APN-Configuration-Profile (AVP 1429)
-        apn_config_profile_context = self.generate_vendor_avp(1423, "c0", 10415, format(int(0),"x").zfill(8))
-        all_apn_config_included = self.generate_vendor_avp(1428, "c0", 10415, format(int(0),"x").zfill(8))
-        apn_config_profile = self.generate_vendor_avp(1429, "c0", 10415, apn_config_profile_context + all_apn_config_included + apn_config)
+            apn_context_id = self.generate_vendor_avp(1423, "c0", 10415, self.int_to_hex(apn_context_identifer_count, 4))
+            apn_service_selection = self.generate_avp(493, "40", self.string_to_hex(str(apn_data['apn'])))
+            apn_pdn_type = self.generate_vendor_avp(1456, "c0", 10415, self.int_to_hex(int(apn_data['ip_version']), 4))
+
+            apn_ambr_ul_avp = self.generate_vendor_avp(516, "c0", 10415, self.int_to_hex(int(apn_data['apn_ambr_ul']), 4))
+            apn_ambr_dl_avp = self.generate_vendor_avp(515, "c0", 10415, self.int_to_hex(int(apn_data['apn_ambr_dl']), 4))
+            apn_ambr = self.generate_vendor_avp(1435, "c0", 10415, apn_ambr_ul_avp + apn_ambr_dl_avp)
+
+            apn_configurations += self.generate_vendor_avp(1430, "c0", 10415,
+                apn_context_id + apn_service_selection + apn_pdn_type + apn_ambr)
+            apn_context_identifer_count += 1
+
+        if not apn_configurations:
+            self.logTool.log(service='HSS', level='error',
+                             message="SWx SAR: IMSI=" + str(imsi) + " apn_list_swx references APN ids that could not be loaded; rejecting with 5450",
+                             redisClient=self.redisMessaging)
+            experimental_result = self.generate_vendor_avp(266, 40, 10415, "")
+            experimental_result += self.generate_avp(298, 40, self.int_to_hex(5450, 4))
+            avp += self.generate_avp(297, 40, experimental_result)
+            response = self.generate_diameter_packet("01", "40", 301, 16777265, packet_vars['hop-by-hop-identifier'], packet_vars['end-to-end-identifier'], avp)
+            return response
 
-        # AMBR (AVP 1435)
-        ambr_ul = self.generate_vendor_avp(516, "c0", 10415, format(int(50000000),"x").zfill(8))
-        ambr_dl = self.generate_vendor_avp(515, "c0", 10415, format(int(100000000),"x").zfill(8))
+        # APN-Configuration-Profile (AVP 1429): use the first allowed APN as default Context-Identifier.
+        apn_config_profile_context = self.generate_vendor_avp(1423, "c0", 10415, self.int_to_hex(1, 4))
+        all_apn_config_included = self.generate_vendor_avp(1428, "c0", 10415, format(int(0),"x").zfill(8))
+        apn_config_profile = self.generate_vendor_avp(1429, "c0", 10415,
+            apn_config_profile_context + all_apn_config_included + apn_configurations)
+
+        # Subscriber UE-AMBR (TS 29.273 keeps the same AMBR encoding as S6a; SWx and S6a should agree).
+        ue_ambr_ul = int(subscriber_details.get('ue_ambr_ul') or 0)
+        ue_ambr_dl = int(subscriber_details.get('ue_ambr_dl') or 0)
+        ambr_ul = self.generate_vendor_avp(516, "c0", 10415, self.int_to_hex(ue_ambr_ul, 4))
+        ambr_dl = self.generate_vendor_avp(515, "c0", 10415, self.int_to_hex(ue_ambr_dl, 4))
         ambr = self.generate_vendor_avp(1435, "c0", 10415, ambr_ul + ambr_dl)
 
         non_3gpp_user_data = self.generate_vendor_avp(1500, "c0", 10415,
@@ -3727,6 +3783,11 @@ def Answer_16777265_301(self, packet_vars, avps):
         avp += non_3gpp_user_data
         avp += self.generate_avp(268, 40, "000007d1")                                                    #Result-Code: DIAMETER_SUCCESS
 
+        self.redisMessaging.sendMetric(serviceName='diameter', metricName='prom_diam_auth_event_count',
+                                        metricType='counter', metricAction='inc', metricValue=1.0,
+                                        metricLabels={"diameter_application_id": 16777265, "diameter_cmd_code": 301, "event": "Success", "imsi_prefix": str(imsi[0:6])},
+                                        metricHelp='Diameter Authentication related Counters',
+                                        metricExpiry=60, usePrefix=True, prefixHostname=self.hostname, prefixServiceName='metric')
         response = self.generate_diameter_packet("01", "40", 301, 16777265, packet_vars['hop-by-hop-identifier'], packet_vars['end-to-end-identifier'], avp)
         return response
 

diff --git a/tests/conftest.py b/tests/conftest.py
@@ -86,6 +86,7 @@ def create_test_db():
         "auc_id": 1,
         "default_apn": "internet",
         "apn_list": "1,2",
+        "apn_list_swx": "1",
         "imsi": test_imsi,
         "msisdn": "100",
     })

diff --git a/tests/db_schema/latest.sql b/tests/db_schema/latest.sql
@@ -220,6 +220,7 @@ CREATE TABLE serving_apn (
 );
 CREATE TABLE subscriber (
 	apn_list VARCHAR(64) NOT NULL,
+	apn_list_swx VARCHAR(64),
 	auc_id INTEGER NOT NULL,
 	default_apn INTEGER NOT NULL,
 	enabled BOOLEAN,

diff --git a/tests/test_API.py b/tests/test_API.py
@@ -172,6 +172,8 @@ def test_A_create_APN_for_Sub(self):
         log.debug("Created APN ID " + str(r.json()['apn_id']))
         self.__class__.template_data['default_apn'] = r.json()['apn_id']
         self.__class__.template_data['apn_list'] = '1,2,' + str(r.json()['apn_id'])
+        # SWx allowlist intentionally narrower than apn_list to verify the per-interface split round-trips through the API
+        self.__class__.template_data['apn_list_swx'] = str(r.json()['apn_id'])
         self.assertEqual(r.status_code, 200, "Status Code should be 200 OK")
 
     def test_A_create_another_APN_for_Sub(self):
@@ -525,6 +527,8 @@ def test_B_GeoRed_MME_create_APN_for_Sub(self):
         self.__class__.apn_id = r.json()['apn_id']
         self.__class__.subscriber_template_data['default_apn'] = r.json()['apn_id']
         self.__class__.subscriber_template_data['apn_list'] = str(r.json()['apn_id'])
+        # No SWx subscription on this fixture; expect apn_list_swx to round-trip as None.
+        self.__class__.subscriber_template_data['apn_list_swx'] = None
         self.assertEqual(r.status_code, 200, "Status Code should be 200 OK")
 
     def test_C_GeoRed_MME_create_Subscriber(self):
@@ -639,6 +643,7 @@ def test_B_GeoRed_PCRF_create_APN_for_Sub(self):
         self.__class__.apn_id = r.json()['apn_id']
         self.__class__.subscriber_template_data['default_apn'] = r.json()['apn_id']
         self.__class__.subscriber_template_data['apn_list'] = str(r.json()['apn_id'])
+        self.__class__.subscriber_template_data['apn_list_swx'] = None
         self.assertEqual(r.status_code, 200, "Status Code should be 200 OK")
 
     def test_C_GeoRed_PCRF_create_Subscriber(self):