Fix host allowlist to require DNS-suffix match

src/graph_notebook/neptune/client.py

@@ -122,9 +122,24 @@
 STREAM_RDF = 'RDF'
 STREAM_ENDPOINTS = {STREAM_PG: 'gremlin', STREAM_RDF: 'sparql'}
 
-ANALYTICS_CONFIG_HOST_IDENTIFIERS = ["neptune-graph", "api.aws", "on.aws", "aws.dev"]
-NEPTUNE_CONFIG_HOST_IDENTIFIERS = ["neptune.amazonaws.com", "neptune.*.amazonaws.com.cn",
-                                   "sc2s.sgov.gov", "c2s.ic.gov"] + ANALYTICS_CONFIG_HOST_IDENTIFIERS
+# Allowed Neptune host suffixes. Entries are matched as case-insensitive DNS
+# suffixes (either an exact match, or a proper subdomain), NOT as substrings
+# or regex patterns. Do NOT add regex metacharacters here -- values are
+# treated as literal domain suffixes.
+ANALYTICS_CONFIG_HOST_IDENTIFIERS = [
+    "neptune-graph.amazonaws.com",
+    "api.aws",
+    "on.aws",
+    "aws.dev",
+]
+
+NEPTUNE_CONFIG_HOST_IDENTIFIERS = [
+    "neptune.amazonaws.com",
+    "neptune.amazonaws.com.cn",
+    "neptune.*.amazonaws.com.cn",
+    "sc2s.sgov.gov",
+    "c2s.ic.gov",
+] + ANALYTICS_CONFIG_HOST_IDENTIFIERS
 
 false_str_variants = [False, 'False', 'false', 'FALSE']
 
@@ -185,13 +200,68 @@
 
 TRAVERSAL_DIRECTIONS = ['both', 'inbound', 'outbound']
 
-def is_allowed_neptune_host(hostname: str, host_allowlist: list):
-    for host_snippet in host_allowlist:
-        if re.search(host_snippet, hostname):
+def is_allowed_neptune_host(hostname: str, host_allowlist: list) -> bool:
+    """
+    Return True iff ``hostname`` matches one of the entries in
+    ``host_allowlist`` as a case-insensitive DNS-label suffix.
+
+    Entries are literal DNS suffixes; a label of ``*`` matches exactly
+    one DNS label of the hostname (not a regex). 
+    """
+    # Reject non-strings and URL-shaped / whitespace-bearing inputs.
+    if not isinstance(hostname, str) or any(c in hostname for c in "/@?# \t\r\n"):
+        return False
+
+    # Normalize and split the hostname into DNS labels.
+    host_labels = hostname.strip().rstrip('.').lower().split('.')
+
+    if any(not label for label in host_labels):
+        return False
+
+    for entry in host_allowlist:
+        # Skip empty / non-string entries so they can't act as a wildcard.
+        if not isinstance(entry, str):
+            continue
+
+        pattern = entry.strip().rstrip('.').lower().split('.')
+        if any(not label for label in pattern):
+            continue
+
+        if _labels_match_tail(pattern, host_labels):
             return True
+
     return False
 
 
+def _labels_match_tail(pattern: list, host_labels: list) -> bool:
+    """
+    Return True if ``pattern`` matches the tail of ``host_labels``
+    label-by-label. A ``*`` in ``pattern`` matches any single
+    non-empty label.
+
+    Examples:
+        1. pattern=[neptune, amazonaws, com]
+           host   =[neptune, amazonaws, com]                    -> True  (exact)
+
+        2. pattern=[neptune, amazonaws, com]
+           host   =[my, cluster, neptune, amazonaws, com]       -> True  (subdomain)
+
+        3. pattern=[neptune, *, amazonaws, com, cn]
+           host   =[..., neptune, cn-north-1, amazonaws, com, cn]
+                                                             -> True  (`*` matches 'cn-north-1')
+        4. pattern=[neptune, amazonaws, com]
+           host   =[xneptune, amazonaws, com]                   -> False (label boundary)
+
+        5. pattern=[neptune, *, amazonaws, com, cn]
+           host   =[neptune, amazonaws, com]                    -> False (host too short)
+    """
+    if len(host_labels) < len(pattern):
+        return False
+    tail = host_labels[-len(pattern):]
+    return all(pattern_label == '*' or pattern_label == host_label
+               for pattern_label, host_label in zip(pattern, tail))
+
+
 def get_gremlin_serializer_driver_class(serializer_str: str):
     if serializer_str == GRAPHBINARYV1:
         return serializer.GraphBinarySerializersV1()

test/unit/neptune/__init__.py

@@ -0,0 +1,4 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: Apache-2.0
+"""

test/unit/neptune/test_client.py

@@ -0,0 +1,207 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: Apache-2.0
+"""
+
+import unittest
+
+from graph_notebook.neptune.client import (
+    ANALYTICS_CONFIG_HOST_IDENTIFIERS,
+    NEPTUNE_CONFIG_HOST_IDENTIFIERS,
+    is_allowed_neptune_host,
+)
+
+
+class TestIsAllowedNeptuneHost(unittest.TestCase):
+
+    # ----- Positive cases: real Neptune endpoints must be accepted -----
+
+    def test_commercial_cluster_endpoint(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "my-cluster.cluster-abc123.us-east-1.neptune.amazonaws.com",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_commercial_reader_endpoint(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "my-cluster.cluster-ro-abc123.us-east-1.neptune.amazonaws.com",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_commercial_instance_endpoint(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "my-instance.abc123.us-east-1.neptune.amazonaws.com",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_china_cluster_endpoint(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "my-cluster.cluster-abc123.cn-north-1.neptune.amazonaws.com.cn",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_china_northwest_cluster_endpoint(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "my-cluster.cluster-abc123.cn-northwest-1.neptune.amazonaws.com.cn",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_analytics_endpoint(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "g-abcdef.g-abcdef123.us-east-1.neptune-graph.amazonaws.com",
+            ANALYTICS_CONFIG_HOST_IDENTIFIERS))
+
+    def test_c2s_endpoint(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "my-cluster.cluster-abc.us-iso-east-1.c2s.ic.gov",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_sc2s_endpoint(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "my-cluster.cluster-abc.us-isob-east-1.sc2s.sgov.gov",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_bare_suffix_matches_itself(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "neptune.amazonaws.com", NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_case_insensitive(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "MyCluster.Cluster-Abc.US-EAST-1.Neptune.AmazonAWS.Com",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_trailing_dot_is_normalized(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "my-cluster.cluster-abc.us-east-1.neptune.amazonaws.com.",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    # ----- Regression cases: attacker bypasses must be rejected -----
+
+    def test_rejects_original_poc_bypass(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "neptune-graph-heli9.requestcatcher.com",
+            ANALYTICS_CONFIG_HOST_IDENTIFIERS))
+
+    def test_rejects_neptune_graph_prefix_bypass(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "neptune-graph.attacker.example",
+            ANALYTICS_CONFIG_HOST_IDENTIFIERS))
+
+    def test_rejects_neptune_suffix_in_middle(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "neptune.amazonaws.com.evil.example",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_rejects_lookalike_without_dot_boundary(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "xneptune.amazonaws.com", NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_rejects_hyphen_prefix_lookalike(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "xneptune-graph.amazonaws.com",
+            ANALYTICS_CONFIG_HOST_IDENTIFIERS))
+
+    def test_rejects_dot_wildcard_bypass_from_old_regex(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "neptuneXamazonawsYcomZcn", NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_rejects_amazonaws_lookalike_com_cn(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "neptune-x.myamazonaws.com.cn",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_rejects_url_shaped_input(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "https://neptune.amazonaws.com/",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_rejects_input_with_userinfo(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "attacker@neptune.amazonaws.com",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_rejects_input_with_query_string(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "neptune.amazonaws.com?x=1",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_rejects_input_with_fragment(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "neptune.amazonaws.com#frag",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_rejects_empty_hostname(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "", NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_rejects_none_hostname(self):
+        self.assertFalse(is_allowed_neptune_host(
+            None, NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_rejects_non_string_hostname(self):
+        self.assertFalse(is_allowed_neptune_host(
+            12345, NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_empty_allowlist_rejects_everything(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "my-cluster.us-east-1.neptune.amazonaws.com", []))
+
+    # ----- Custom allowlist handling -----
+
+    def test_custom_allowlist_accepts_subdomain(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "foo.internal.example.com", ["internal.example.com"]))
+
+    def test_custom_allowlist_rejects_lookalike(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "internal.example.com.evil.tld", ["internal.example.com"]))
+
+    def test_custom_allowlist_ignores_empty_entries(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "attacker.example.com", ["", None, "  "]))
+
+    # ----- Single-label wildcard (``*``) handling -----
+
+    def test_wildcard_matches_legacy_china_form(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "instance.cluster.neptune.cn-north-1.amazonaws.com.cn",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_wildcard_matches_northwest_legacy_china_form(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "instance.cluster.neptune.cn-northwest-1.amazonaws.com.cn",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_wildcard_matches_pattern_directly(self):
+        self.assertTrue(is_allowed_neptune_host(
+            "neptune.evil.amazonaws.com.cn",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_wildcard_consumes_exactly_one_label(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "neptune.foo.bar.amazonaws.com.cn",
+            ["neptune.*.amazonaws.com.cn"]))
+        self.assertFalse(is_allowed_neptune_host(
+            "neptune.amazonaws.com.cn",
+            ["neptune.*.amazonaws.com.cn"]))
+
+    def test_wildcard_rejects_prefix_lookalike(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "evilneptune.foo.amazonaws.com.cn",
+            ["neptune.*.amazonaws.com.cn"]))
+
+    def test_wildcard_does_not_match_wrong_tail(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "neptune.foo.amazonaws.com",
+            ["neptune.*.amazonaws.com.cn"]))
+        self.assertFalse(is_allowed_neptune_host(
+            "neptune.foo.evilamazonaws.com.cn",
+            ["neptune.*.amazonaws.com.cn"]))
+
+    def test_rejects_empty_label(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "neptune..amazonaws.com.cn", NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+    def test_wildcard_still_rejects_original_poc(self):
+        self.assertFalse(is_allowed_neptune_host(
+            "neptune-graph-heli9.requestcatcher.com",
+            NEPTUNE_CONFIG_HOST_IDENTIFIERS))
+
+
+if __name__ == '__main__':
+    unittest.main()