feat: implement inbound roaming reject

CHANGELOG.md

@@ -19,6 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Building PyHSS with `python3 -m build` and as debian package.
 - RAT restriction checking for subscribers.
 - Automatic database upgrades (from 1.0.1 or higher).
+- Allow users to configure `hss.roaming.inbound.reject_unknown_imsis_with` to either `IMSI_UNKNOWN` (default) or `ROAMING_NOT_ALLOWED`.
 
 ### Changed
 

config.yaml

@@ -79,8 +79,17 @@ hss:
     - 'scscf.ims.mnc001.mcc001.3gppnetwork.org'
 
   roaming:
+    inbound:
+      # How to handle unknown IMSIs when inbound roaming. Valid options are:
+      #   IMSI_UNKNOWN - Reject the request with the "IMSI_UNKNOWN" error code.
+      #   ROAMING_NOT_ALLOWED - Reject the request with the "ROAMING_NOT_ALLOWED" error code.
+      # The latter is useful when operating a private network and public SIMs should not be allowed to connect.
+      # "Roaming not allowed" will cause the UE to not try again on this network.
+      # "Imsi unknown" may cause the UE to believe that its subscription is invalid.
+      # For private networks, "ROAMING_NOT_ALLOWED" is recommended. Otherwise, use "IMSI_UNKNOWN".
+      reject_unknown_imsis_with: "ROAMING_NOT_ALLOWED"
     outbound:
-      # Whether or not to a subscriber to connect to an undefined network when outbound roaming.
+      # Whether or not to allow a subscriber to connect to an undefined network when outbound roaming.
       allow_undefined_networks: True
 
   # SCTP Socket Parameters

docker/.env

@@ -26,6 +26,7 @@ HSS_DSR_EXTERNAL_IDENTIFIER=example
 HSS_IGNORE_PURGE_UE_REQUEST=False
 HSS_SCSCF_POOL=scscf.ims.mnc001.mcc001.3gppnetwork.org
 HSS_ROAMING_ALLOW_UNDEFINED_NETWORKS=True
+HSS_ROAMING_REJECT_UNKNOWN_IMSIS_WITH="ROAMING_NOT_ALLOWED"
 HSS_SCTP_RTO_MAX=5000
 HSS_SCTP_RTO_MIN=500
 HSS_SCTP_RTO_INITIAL=1000

docker/config.yaml

@@ -79,8 +79,17 @@ hss:
     - "${HSS_SCSCF_POOL:-scscf.ims.mnc001.mcc001.3gppnetwork.org}"
 
   roaming:
+    inbound:
+      # How to handle unknown IMSIs when inbound roaming. Valid options are:
+      #   IMSI_UNKNOWN - Reject the request with the "IMSI_UNKNOWN" error code.
+      #   ROAMING_NOT_ALLOWED - Reject the request with the "ROAMING_NOT_ALLOWED" error code.
+      # The latter is useful when operating a private network and public SIMs should not be allowed to connect.
+      # "Roaming not allowed" will cause the UE to not try again on this network.
+      # "Imsi unknown" may cause the UE to believe that its subscription is invalid.
+      # For private networks, "ROAMING_NOT_ALLOWED" is recommended. Otherwise, use "IMSI_UNKNOWN".
+      reject_unknown_imsis_with: "${HSS_ROAMING_REJECT_UNKNOWN_IMSIS_WITH:-ROAMING_NOT_ALLOWED}"
     outbound:
-      # Whether or not to a subscriber to connect to an undefined network when outbound roaming.
+      # Whether or not to allow a subscriber to connect to an undefined network when outbound roaming.
       allow_undefined_networks: ${HSS_ROAMING_ALLOW_UNDEFINED_NETWORKS:-True}
 
   # SCTP Socket Parameters

lib/diameter.py

@@ -130,6 +130,13 @@ def __init__(
 
         ]
 
+    @staticmethod
+    def get_unknown_imsi_reject_cause() -> int:
+        if config.get('hss', {}).get('roaming', {}).get('inbound', {}).get('reject_unknown_imsis_with', 'IMSI_UNKNOWN') == 'ROAMING_NOT_ALLOWED':
+            return 5004 # DIAMETER_ERROR_ROAMING_NOT_ALLOWED
+        return 5001 # DIAMETER_ERROR_USER_UNKNOWN
+
+
     #Generates rounding for calculating padding
     def myround(self, n, base=4):
         if(n > 0):
@@ -2249,7 +2256,7 @@ def Answer_16777251_318(self, packet_vars, avps):
                                             usePrefix=True, 
                                             prefixHostname=self.hostname, 
                                             prefixServiceName='metric')
-            #Handle if the subscriber is not present in HSS return "DIAMETER_ERROR_USER_UNKNOWN"
+            #Handle if the subscriber is not present in HSS return the appropriate error
             self.logTool.log(service='HSS', level='debug', message="Subscriber " + str(imsi) + " is unknown in database", redisClient=self.redisMessaging)
             avp = ''
             session_id = self.get_avp_data(avps, 263)[0]                                                     #Get Session-ID
@@ -2260,7 +2267,7 @@ def Answer_16777251_318(self, packet_vars, avps):
             #Experimental Result AVP(Response Code for Failure)
             avp_experimental_result = ''
             avp_experimental_result += self.generate_vendor_avp(266, 40, 10415, '')                         #AVP Vendor ID
-            avp_experimental_result += self.generate_avp(298, 40, self.int_to_hex(5001, 4))                 #AVP Experimental-Result-Code: DIAMETER_ERROR_USER_UNKNOWN (5001)
+            avp_experimental_result += self.generate_avp(298, 40, self.int_to_hex(self.get_unknown_imsi_reject_cause(), 4))                 #AVP Experimental-Result-Code: DIAMETER_ERROR_USER_UNKNOWN (5001) or DIAMETER_ERROR_ROAMING_NOT_ALLOWED (5004)
             avp += self.generate_avp(297, 40, avp_experimental_result)                                      #AVP Experimental-Result(297)
 
             avp += self.generate_avp(277, 40, "00000001")                                                    #Auth-Session-State

lib/gsup/controller/air.py

@@ -13,6 +13,7 @@
 from logtool import LogTool
 from utils import validate_imsi, InvalidIMSI
 
+from pyhss_config import config
 
 class AIRController(GsupController):
     def __init__(self, logger: LogTool, database: Database):
@@ -27,6 +28,13 @@ def get_num_vectors_req(self, message: dict):
             return max_num
         return ret
 
+    @staticmethod
+    def _get_unknown_subscriber_reject_cause() -> GMMCause:
+        if config.get('hss', {}).get('roaming', {}).get('inbound', {}).get('reject_unknown_imsis_with', 'IMSI_UNKNOWN') == 'ROAMING_NOT_ALLOWED':
+            return GMMCause.ROAMING_NOTALLOWED
+        else:
+            return GMMCause.IMSI_UNKNOWN
+
     async def handle_message(self, peer: IPAPeer, message: GsupMessage):
         request_dict = message.to_dict()
         imsi = GsupMessageUtil.get_first_ie_by_name(GsupMessageUtil.GSUP_MSG_IE_IMSI, request_dict)
@@ -79,7 +87,7 @@ async def handle_message(self, peer: IPAPeer, message: GsupMessage):
                 peer,
                 GsupMessageBuilder().with_msg_type(MsgType.SEND_AUTH_INFO_ERROR)
                 .with_ie('imsi', imsi)
-                .with_ie('cause', GMMCause.IMSI_UNKNOWN.value)
+                .with_ie('cause', self._get_unknown_subscriber_reject_cause().value)
                 .build(),
             )
         except Exception as e:

lib/pyhss_config.py

@@ -35,4 +35,19 @@ def load_config():
     sys.exit(1)
 
 
+def validate_config():
+    """Validate configuration values at startup.
+
+    Refuses to start if config options have unexpected values,
+    preventing silent misconfiguration. Add future validations here."""
+
+    valid_reject_causes = {"IMSI_UNKNOWN", "ROAMING_NOT_ALLOWED"}
+    reject_cause = config.get('hss', {}).get('roaming', {}).get('inbound', {}).get('reject_unknown_imsis_with', 'IMSI_UNKNOWN')
+    if reject_cause not in valid_reject_causes:
+        print(f"ERROR: invalid value for hss.roaming.inbound.reject_unknown_imsis_with: '{reject_cause}'. "
+              f"Valid options are: {', '.join(sorted(valid_reject_causes))}")
+        sys.exit(1)
+
+
 load_config()
+validate_config()

tests/config.yaml

@@ -30,6 +30,8 @@ hss:
     - 'scscf.ims.mnc001.mcc001.3gppnetwork.org'
 
   roaming:
+    inbound:
+      reject_unknown_imsis_with: "ROAMING_NOT_ALLOWED"
     outbound:
       allow_undefined_networks: True
 

tests/test_pyhss_config.py

@@ -0,0 +1,54 @@
+# Copyright 2025 Lennart Rosam <hello@takuto.de>
+# SPDX-License-Identifier: AGPL-3.0-or-later
+from unittest import TestCase
+
+from pyhss_config import config, validate_config
+
+
+class ConfigValidationTest(TestCase):
+    def setUp(self):
+        import pyhss_config
+        self._saved_config = dict(pyhss_config.config) if pyhss_config.config else {}
+
+    def tearDown(self):
+        import pyhss_config
+        pyhss_config.config = self._saved_config
+
+    def test_valid_reject_causes(self):
+        """Valid reject cause values should pass validation."""
+        import pyhss_config
+        for value in ('IMSI_UNKNOWN', 'ROAMING_NOT_ALLOWED'):
+            with self.subTest(value=value):
+                pyhss_config.config = {
+                    'hss': {
+                        'roaming': {
+                            'inbound': {
+                                'reject_unknown_imsis_with': value
+                            }
+                        }
+                    }
+                }
+                validate_config()
+
+    def test_config_passes_with_empty_or_nested_missing(self):
+        """Config should pass validation when nested keys are missing."""
+        import pyhss_config
+        for cfg in ({}, {'hss': {}}, {'hss': {'roaming': {}}}, {'hss': {'roaming': {'inbound': {}}}}):
+            with self.subTest(config=cfg):
+                pyhss_config.config = cfg
+                validate_config()
+
+    def test_invalid_value_raises(self):
+        """Invalid value should cause sys.exit."""
+        import pyhss_config
+        pyhss_config.config = {
+            'hss': {
+                'roaming': {
+                    'inbound': {
+                        'reject_unknown_imsis_with': 'INVALID_CAUSE'
+                    }
+                }
+            }
+        }
+        with self.assertRaises(SystemExit):
+            validate_config()