hostcfgd: Validate RADIUS server fields before creating PAM files

scripts/hostcfgd

@@ -8,6 +8,7 @@ import subprocess
 import syslog
 import signal
 import re
+import socket
 import jinja2
 import psutil
 import time
@@ -97,6 +98,60 @@ RADIUS_SERVER_AUTH_TYPE_DEFAULT = "pap"
 RADIUS_PAM_AUTH_CONF_DIR = "/etc/pam_radius_auth.d/"
 RADIUS_SERVER_SKIP_MSG_AUTH = False
 
+
+def _radius_host_valid(host):
+    if not isinstance(host, str) or not host or any(ord(char) < 33 or ord(char) == 127 for char in host):
+        return False
+
+    try:
+        if '%' in host:
+            address, scope = host.split('%', 1)
+            if '%' in scope or not re.fullmatch(r'[A-Za-z0-9_.-]{1,64}', scope):
+                return False
+            ipaddress.IPv6Address(address)
+        else:
+            ipaddress.ip_address(host)
+        return True
+    except ValueError:
+        if '%' in host:
+            return False
+
+    # Reject non-canonical IPv4 forms before applying DNS hostname syntax.
+    try:
+        socket.inet_aton(host)
+        return False
+    except OSError:
+        pass
+
+    hostname = host[:-1] if host.endswith('.') else host
+    if not hostname or len(hostname) > 253:
+        return False
+    label_pattern = re.compile(r'[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?')
+    return all(label_pattern.fullmatch(label) for label in hostname.split('.'))
+
+
+def radius_server_entry_valid(server):
+    try:
+        if not _radius_host_valid(server['ip']):
+            raise ValueError('invalid host')
+        if not isinstance(server['auth_port'], str) or not re.fullmatch(r'[0-9]+', server['auth_port']):
+            raise ValueError('invalid auth_port')
+        auth_port = int(server['auth_port'])
+        if not 1 <= auth_port <= 65535:
+            raise ValueError('auth_port out of range')
+        server['auth_port'] = str(auth_port)
+        return True
+    except (KeyError, ValueError) as error:
+        syslog.syslog(
+            syslog.LOG_ERR,
+            'RADIUS_SERVER: skipping invalid server entry '
+            '(ip={!r}, auth_port={!r}): {}'.format(
+                server.get('ip'), server.get('auth_port'), error
+            )
+        )
+        return False
+
+
 # FIPS
 FIPS_CONFIG_FILE = '/etc/sonic/fips.json'
 OPENSSL_FIPS_CONFIG_FILE = '/etc/fips/fips_enable'
@@ -699,6 +754,8 @@ class AaaCfg(object):
             "RADIUS_SERVER|{}: src_intf has no usable IP addr.".format(addr))
                         del server['src_ip']
 
+                if not radius_server_entry_valid(server):
+                    continue
                 radsrvs_conf.append(server)
             radsrvs_conf = sorted(radsrvs_conf, key=lambda t: int(t['priority']), reverse=True)
 

tests/hostcfgd/hostcfgd_radius_test.py

@@ -10,6 +10,7 @@
 from unittest import TestCase, mock
 from tests.hostcfgd.test_radius_vectors import HOSTCFGD_TEST_RADIUS_VECTOR
 from tests.common.mock_configdb import MockConfigDb, MockDBConnector
+from tests.common.mock_restart_waiter import MockRestartWaiter
 from sonic_py_common.general import getstatusoutput_noshell
 
 
@@ -33,6 +34,8 @@
 hostcfgd.ConfigDBConnector = MockConfigDb
 hostcfgd.DBConnector = MockDBConnector
 hostcfgd.Table = mock.Mock()
+hostcfgd.swsscommon.RestartWaiter = MockRestartWaiter
+
 
 class TestHostcfgdRADIUS(TestCase):
     """
@@ -42,6 +45,62 @@ def run_diff(self, file1, file2):
         _, output = getstatusoutput_noshell(['diff', '-ur', file1, file2])
         return output
 
+    @parameterized.expand([
+        ('ipv4', {'ip': '192.0.2.1', 'auth_port': '1812'}, True),
+        ('ipv6', {'ip': '2001:db8::1', 'auth_port': '1812'}, True),
+        ('scoped_ipv6', {'ip': 'fe80::1%Ethernet0', 'auth_port': '1812'}, True),
+        ('hostname', {'ip': 'radius.example.com', 'auth_port': '1812'}, True),
+        ('hostname_with_numeric_label',
+         {'ip': '123.radius.example.com', 'auth_port': '1812'}, True),
+        ('noncanonical_ipv4_two_part',
+         {'ip': '192.0.513', 'auth_port': '1812'}, False),
+        ('noncanonical_ipv4_integer',
+         {'ip': '3221225985', 'auth_port': '1812'}, False),
+        ('invalid_host', {'ip': 'invalid_host', 'auth_port': '1812'}, False),
+        ('invalid_port', {'ip': '192.0.2.1', 'auth_port': 'invalid'}, False),
+        ('port_with_newline', {'ip': '192.0.2.1', 'auth_port': '1812\n'}, False),
+        ('scope_with_newline', {'ip': 'fe80::1%Ethernet0\n', 'auth_port': '1812'}, False),
+        ('port_too_low', {'ip': '192.0.2.1', 'auth_port': '0'}, False),
+        ('port_too_high', {'ip': '192.0.2.1', 'auth_port': '65536'}, False),
+    ])
+    def test_radius_server_entry_valid(self, _, server, expected):
+        with mock.patch.object(hostcfgd.syslog, 'syslog'):
+            self.assertEqual(hostcfgd.radius_server_entry_valid(server), expected)
+
+    def test_radius_server_port_is_normalized(self):
+        server = {'ip': '192.0.2.1', 'auth_port': '01812'}
+        self.assertTrue(hostcfgd.radius_server_entry_valid(server))
+        self.assertEqual(server['auth_port'], '1812')
+
+    def test_radius_hostname_is_rendered_in_pam_config(self):
+        server = {
+            'ip': 'radius.example.com',
+            'auth_port': '1812',
+            'auth_type': 'pap',
+            'retransmit': '3',
+            'timeout': '5',
+            'passkey': 'shared-secret',
+            'skip_msg_auth': False,
+        }
+        self.assertTrue(hostcfgd.radius_server_entry_valid(server))
+
+        env = hostcfgd.jinja2.Environment(loader=hostcfgd.jinja2.FileSystemLoader('/'))
+        env.filters['sub'] = hostcfgd.sub
+        common_auth = env.get_template(
+            os.path.abspath(templates_path + '/common-auth-sonic.j2')
+        ).render(
+            debug=False,
+            trace=False,
+            auth={'login': 'radius,local'},
+            servers=[server],
+        )
+        server_config = env.get_template(
+            os.path.abspath(templates_path + '/pam_radius_auth.conf.j2')
+        ).render(server=server)
+
+        self.assertIn('/radius.example.com_1812.conf', common_auth)
+        self.assertIn('[radius.example.com]:1812', server_config)
+
 
     @parameterized.expand(HOSTCFGD_TEST_RADIUS_VECTOR)
     def test_hostcfgd_radius(self, test_name, test_data):

tests/hostcfgd/test_radius_vectors.py

@@ -91,6 +91,15 @@
                         "timeout": "1",
                         "passkey": "pass1",
                         "skip_msg_auth": "true",
+                    },
+                    "invalid_host": {
+                        "auth_type": "pap",
+                        "passkey": "invalid1",
+                    },
+                    "10.10.10.6": {
+                        "auth_port": "65536",
+                        "auth_type": "pap",
+                        "passkey": "invalid2",
                     }
                 },
             },