Fix ntp whitelist

modules/test/ntp/conf/module_config.json

@@ -25,81 +25,6 @@
         "name": "ntp.network.ntp_dhcp",
         "test_description": "Accept NTP address over DHCP or from public trusted sources",
         "expected_behavior": "Device can accept NTP server address, provided by the DHCP server (DHCP OFFER PACKET) or from public trusted sources.",
-        "config": {
-          "pools_with_subdomains": [
-            "pool.ntp.org",
-            "europe.pool.ntp.org",
-            "asia.pool.ntp.org",
-            "north-america.pool.ntp.org",
-            "south-america.pool.ntp.org",
-            "oceania.pool.ntp.org",
-            "africa.pool.ntp.org",
-            "us.pool.ntp.org",
-            "de.pool.ntp.org",
-            "fr.pool.ntp.org",
-            "uk.pool.ntp.org",
-            "jp.pool.ntp.org",
-            "cn.pool.ntp.org",
-            "br.pool.ntp.org",
-            "in.pool.ntp.org",
-            "za.pool.ntp.org",
-            "au.pool.ntp.org",
-            "ca.pool.ntp.org",
-            "ch.pool.ntp.org",
-            "it.pool.ntp.org",
-            "es.pool.ntp.org",
-            "pl.pool.ntp.org",
-            "nl.pool.ntp.org",
-            "se.pool.ntp.org",
-            "fi.pool.ntp.org",
-            "dk.pool.ntp.org",
-            "no.pool.ntp.org",
-            "be.pool.ntp.org",
-            "pt.pool.ntp.org",
-            "tr.pool.ntp.org",
-            "gr.pool.ntp.org",
-            "cz.pool.ntp.org",
-            "sk.pool.ntp.org",
-            "hu.pool.ntp.org",
-            "ro.pool.ntp.org",
-            "bg.pool.ntp.org",
-            "lt.pool.ntp.org",
-            "lv.pool.ntp.org",
-            "ee.pool.ntp.org",
-            "ua.pool.ntp.org",
-            "rs.pool.ntp.org",
-            "hr.pool.ntp.org",
-            "si.pool.ntp.org",
-            "ntp.ubuntu.pool.ntp.org",
-            "debian.pool.ntp.org"
-          ],
-          "single_servers": [
-            "ntp.ubuntu.com",
-            "ntp.debian.org",
-            "time.google.com",
-            "time.cloudflare.com",
-            "time.windows.com",
-            "time.apple.com",
-            "time.nist.gov",
-            "utcnist.colorado.edu",
-            "ntp1.ptb.de",
-            "ntp2.ptb.de",
-            "ntp3.ptb.de",
-            "ntp1.npl.co.uk",
-            "ntp2.npl.co.uk",
-            "ntp.nict.jp",
-            "ntp1.nict.jp",
-            "ntp2.nict.jp",
-            "ntp1.t-online.de",
-            "ntp2.t-online.de",
-            "ntp3.t-online.de",
-            "ntp4.t-online.de"
-          ],
-          "dns_servers": [
-            "8.8.8.8",
-            "1.1.1.1"
-          ]
-        },
         "recommendations": [
           "Install an NTP client that supports fetching the NTP servers from DHCP options",
           "Verify that the device is configured to accept DHCP Option 42",

modules/test/ntp/python/requirements.txt

@@ -5,4 +5,5 @@
 # User defined packages
 scapy==2.7.0
 pyshark==0.6
-dnspython==2.8.0
+aiohttp==3.13.5
+ntplib==0.4.0

modules/test/ntp/python/src/ntp_module.py

@@ -19,7 +19,7 @@
 from collections import defaultdict
 from jinja2 import Environment, FileSystemLoader
 import pyshark
-from ntp_white_list import NTPWhitelistResolver
+from ntp_white_list import check_all_ips
 
 LOG_NAME = 'test_ntp'
 MODULE_REPORT_FILE_NAME = 'ntp_report.j2.html'
@@ -319,16 +319,8 @@ def _ntp_network_ntp_support(self):
     LOGGER.info(result[1])
     return result
 
-  def _ntp_network_ntp_dhcp(self, config):
+  def _ntp_network_ntp_dhcp(self):
     LOGGER.info('Running ntp.network.ntp_dhcp')
-    try:
-      ntp_whitelist_resolver = NTPWhitelistResolver(
-          config=config,
-          logger=LOGGER
-      )
-    except Exception as e:
-      LOGGER.error(f'Error initializing NTPWhitelistResolver: {e}')
-      return 'Error', 'Failed to initialize NTP whitelist resolver'
 
     # Read the pcap files
     packet_capture = (rdpcap(self.startup_capture_file) +
@@ -360,27 +352,29 @@ def _ntp_network_ntp_dhcp(self, config):
           LOGGER.info('Device sent NTP request to non-DHCP provided NTP server')
           ntp_to_remote = True
           ntp_to_remote_ips.add(dest_ip)
+    ips_trusted = []
+    all_ips_trusted = False
+    if ntp_to_remote_ips:
+      ips_trusted =  check_all_ips(list(ntp_to_remote_ips))
+      LOGGER.debug(f'Checked NTP remote IPs: {ips_trusted}')
+      all_ips_trusted = all(is_trusted for _, is_trusted in ips_trusted)
 
-    ntp_to_remote_trusted = bool(ntp_to_remote_ips) and all(
-        ntp_whitelist_resolver.is_ip_whitelisted(ip) for ip in ntp_to_remote_ips
-    )
 
     result_details = [f'NTP request to {self._ntp_server}']
 
-    for ip in ntp_to_remote_ips:
-      if ntp_whitelist_resolver.is_ip_whitelisted(ip):
-        LOGGER.info(f'NTP server {ip} is in the trusted whitelist')
+    for ip, is_trusted in ips_trusted:
+      if is_trusted:
+        LOGGER.info(f'NTP server {ip} is trusted')
       else:
-        LOGGER.info(f'NTP server {ip} is NOT in the trusted whitelist')
+        LOGGER.info(f'NTP server {ip} is NOT trusted')
       result_details.append(f'NTP request to {ip}')
 
     result_state = 'Feature Not Detected'
     result_message = 'Device has not sent any NTP requests'
 
-
     if device_sends_ntp:
       if ntp_to_local and ntp_to_remote:
-        if ntp_to_remote_trusted:
+        if all_ips_trusted:
           result_state = True
           result_message = ('Device sent NTP request to DHCP provided ' +
                           'server and trusted non-DHCP provided servers')
@@ -389,7 +383,7 @@ def _ntp_network_ntp_dhcp(self, config):
           result_message = ('Device sent NTP request to DHCP provided ' +
                            'server and to untrusted non-DHCP provided server')
       elif ntp_to_remote:
-        if ntp_to_remote_trusted:
+        if all_ips_trusted:
           result_state = False
           result_message = ('Device sent NTP request to trusted ' +
                            'non-DHCP provided server')
@@ -404,5 +398,4 @@ def _ntp_network_ntp_dhcp(self, config):
     if not ntp_to_local:
       result_details.pop(0)
 
-    LOGGER.info(result_state)
     return result_state, result_message, result_details

modules/test/ntp/python/src/ntp_white_list.py

@@ -1,118 +1,65 @@
 """Module to resolve NTP whitelist domains to IP addresses asynchronously."""
 
 import asyncio
+import aiohttp
 import concurrent.futures
-import dns.asyncresolver
-from logging import Logger
+import ntplib
 
+NTP_URL = 'https://ntppool.org/scores/{ip}/json'
 
-class NTPWhitelistResolver:
-  """Class to resolve NTP whitelist domains to IP addresses."""
 
-  def __init__(self,
-               config: dict,
-               logger: Logger,
-               semaphore_limit: int = 50,
-               timeout: int = 30
-              ):
-    self.config = config
-    self.semaphore_limit = semaphore_limit
-    self.timeout = timeout
-    self._logger = logger
-    self._ip_whitelist = self._get_ntp_whitelist_ips()
+async def fetch_ntp_status(
+    session: aiohttp.ClientSession,
+    ip: str
+  ) -> tuple[str, bool]:
+  """Fetch the NTP status for a single IP address."""
+  try:
+    async with session.get(NTP_URL.format(ip=ip), timeout=10) as response:
+      if response.status == 200:
+        data = await response.json()
+        score = data['monitors'][0].get('score', 0)
+        active = score >= 10
+        return (ip, active)
+      else:
+        return (ip, False)
+  except Exception:
+    return (ip, False)
 
-  # Create the final list of NTP domains to resolve
-  def _create_final_ntp_domain_list(
-    self,
-    pools_with_subdomains: list[str],
-    single_servers: list[str]
-  ) -> list[str]:
-    ntp_domains = []
-    for pool in pools_with_subdomains:
-      for i in range(4):
-        ntp_domains.append(f"{i}.{pool}")
-      ntp_domains.append(pool)
-    ntp_domains.extend(single_servers)
-    return ntp_domains
 
-  # Resolve a domain to its IP addresses
-  async def _resolve_domain(
-      self, domain: str, dns_servers: list[str], attempts: int = 2
-  ) -> set[str]:
-    ips = set()
-    for _ in range(attempts):
-      for dns_server in dns_servers:
-        resolver = dns.asyncresolver.Resolver()
-        resolver.nameservers = [dns_server]
-        try:
-          answers = await resolver.resolve(domain, "A", lifetime=2)
-          for rdata in answers:
-            ips.add(rdata.address)
-        except Exception:
-          pass
-        try:
-          answers6 = await resolver.resolve(domain, "AAAA", lifetime=2)
-          for rdata in answers6:
-            ips.add(rdata.address)
-        except Exception:
-          pass
-    return ips
+async def _check_all_ips_async(ip_list: list[str]) -> list[tuple[str, bool]]:
+  """Check NTP status for all IPs asynchronously."""
+  async with aiohttp.ClientSession() as session:
+    tasks = [fetch_ntp_status(session, ip) for ip in ip_list]
+    # Run all tasks concurrently
+    results = await asyncio.gather(*tasks)
+    return results
 
-  async def _sem_task(
-      self,
-      domain: str,
-      dns_servers: list[str],
-      semaphore: asyncio.Semaphore
-  ) -> set[str]:
-    async with semaphore:
-      return await self._resolve_domain(domain, dns_servers)
+def _get_ntp_data(ip):
+  """Check NTP status for a single IP address."""
+  client = ntplib.NTPClient()
+  try:
+    client.request(ip, version=3, timeout=2)
+    return True
+  except ntplib.NTPException:
+    return False
 
-  # Get IPs for NTP whitelist
-  async def _get_ips_whitelist(
-        self, config, semaphore_limit: int, timeout: int
-  ) -> set[str]:
-    pools_with_subdomains = config.get("pools_with_subdomains", [])
-    single_servers = config.get("single_servers", [])
-    dns_servers = config.get("dns_servers", [])
-    ntp_domain_list = self._create_final_ntp_domain_list(
-        pools_with_subdomains, single_servers
-    )
-    semaphore = asyncio.Semaphore(semaphore_limit)
-    tasks = [self._sem_task(
-                domain,
-                dns_servers,
-                semaphore)
-                for domain in ntp_domain_list
-            ]
-    all_ips = set()
-    for coro in asyncio.as_completed(tasks, timeout=timeout):
-      try:
-        ips = await coro
-        if isinstance(ips, set):
-          all_ips.update(ips)
-      except asyncio.TimeoutError:
-        break
-    self._logger.info(f"Added {len(all_ips)} IPs to NTP whitelist.")
-    return all_ips
 
-  def _get_ntp_whitelist_ips(
-      self,
-      semaphore_limit: int = 50,
-      timeout: int = 30
-  ) -> set[str]:
-    # Always run in a separate thread to ensure we have a clean event loop
-    def run_in_thread():
-      new_loop = asyncio.new_event_loop()
-      asyncio.set_event_loop(new_loop)
-      try:
-        return new_loop.run_until_complete(
-            self._get_ips_whitelist(self.config, semaphore_limit, timeout))
-      finally:
-        new_loop.close()
+def check_all_ips(ip_list: list[str]) -> list[tuple[str, bool]]:
+  """Check NTP status for all IPs in a separate thread."""  
+  def run_in_thread():
+    new_loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(new_loop)
+    try:
+      return new_loop.run_until_complete(_check_all_ips_async(ip_list))
+    finally:
+      new_loop.close()
 
-    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
-      return executor.submit(run_in_thread).result()
-
-  # Check if an IP is whitelisted
-  def is_ip_whitelisted(self, ip: str) -> bool:
-    return ip in self._ip_whitelist
+  with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+    future = executor.submit(run_in_thread)
+    ntps = future.result()
+    for i in range(len(ntps)):
+      ip, trusted = ntps[i]
+      if not trusted:
+        trusted = _get_ntp_data(ip)
+      ntps[i] = (ip, trusted)
+    return ntps