Add IPv6 :type and :scope (#3498)

invisig0th · MichaelSquires · web-flow · commit ba1ca2e10967 · 2024-01-23T12:41:13.000-05:00
Co-authored-by: blackout &lt;blackout@vertex.link&gt;
diff --git a/synapse/lib/modelrev.py b/synapse/lib/modelrev.py
@@ -731,6 +731,7 @@ async def revModel_0_2_22(self, layers):
         await self._normFormSubs(layers, 'inet:ipv4', cmprvalu='100.64.0.0/10')
 
     async def revModel_0_2_23(self, layers):
+        await self._normFormSubs(layers, 'inet:ipv6')
         await self._normFormSubsByProp(layers, 'it:sec:cpe', 'v2_2')
 
     async def runStorm(self, text, opts=None):
diff --git a/synapse/models/inet.py b/synapse/models/inet.py
@@ -54,6 +54,35 @@ def getAddrType(ip):
 
     return 'unicast'
 
+# https://en.wikipedia.org/wiki/IPv6_address#Address_scopes
+ipv6_multicast_scopes = {
+    'ff00:': 'reserved',
+    'ff01:': 'interface-local',
+    'ff02:': 'link-local',
+    'ff03:': 'realm-local',
+    'ff04:': 'admin-local',
+    'ff05:': 'site-local',
+    'ff08:': 'organization-local',
+    'ff0e:': 'global',
+    'ff0f:': 'reserved',
+}
+
+scopes_enum = 'reserved,interface-local,link-local,realm-local,admin-local,site-local,organization-local,global,unassigned'
+
+def getAddrScope(ipv6):
+
+    if ipv6.is_loopback:
+        return 'link-local'
+
+    if ipv6.is_link_local:
+        return 'link-local'
+
+    if ipv6.is_multicast:
+        pref = ipv6.compressed[:5]
+        return ipv6_multicast_scopes.get(pref, 'unassigned')
+
+    return 'global'
+
 class Addr(s_types.Str):
 
     def postTypeInit(self):
@@ -579,15 +608,18 @@ def _normPyStr(self, valu):
             v6 = ipaddress.IPv6Address(valu)
             v4 = v6.ipv4_mapped
 
-            subs = {'type': getAddrType(v6)}
+            subs = {
+                'type': getAddrType(v6),
+                'scope': getAddrScope(v6),
+            }
 
             if v4 is not None:
                 v4_int = self.modl.type('inet:ipv4').norm(v4.compressed)[0]
                 v4_str = self.modl.type('inet:ipv4').repr(v4_int)
                 subs['ipv4'] = v4_int
                 return f'::ffff:{v4_str}', {'subs': subs}
 
-            return ipaddress.IPv6Address(valu).compressed, {'subs': subs}
+            return v6.compressed, {'subs': subs}
 
         except Exception as e:
             raise s_exc.BadTypeValu(valu=valu, name=self.name, mesg=str(e)) from None
@@ -2057,6 +2089,13 @@ def getModelDefs(self):
 
                         ('loc', ('loc', {}), {
                             'doc': 'The geo-political location string for the IPv6.'}),
+
+                        ('type', ('str', {}), {
+                            'doc': 'The type of IP address (e.g., private, multicast, etc.).'}),
+
+                        ('scope', ('str', {'enums': scopes_enum}), {
+                            'doc': 'The IPv6 scope of the address (e.g., global, link-local, etc.).'}),
+
                     )),
 
                     ('inet:mac', {}, (
diff --git a/synapse/tests/test_lib_modelrev.py b/synapse/tests/test_lib_modelrev.py
@@ -430,8 +430,8 @@ async def test_modelrev_0_2_22(self):
                 self.eq(node.props.get('type'), 'shared')
 
     async def test_modelrev_0_2_23(self):
-
         async with self.getRegrCore('model-0.2.23') as core:
+            self.len(1, await core.nodes('inet:ipv6="ff01::1" +:type=multicast +:scope=interface-local'))
 
             nodes = await core.nodes('it:sec:cpe:vendor=10web')
             self.len(1, nodes)
diff --git a/synapse/tests/test_model_inet.py b/synapse/tests/test_model_inet.py
@@ -986,11 +986,13 @@ async def test_ipv6(self):
             # Type Tests ======================================================
             t = core.model.type(formname)
 
-            info = {'subs': {'type': 'loopback'}}
+            info = {'subs': {'type': 'loopback', 'scope': 'link-local'}}
             self.eq(t.norm('::1'), ('::1', info))
             self.eq(t.norm('0:0:0:0:0:0:0:1'), ('::1', info))
 
-            info = {'subs': {'type': 'private'}}
+            self.eq(t.norm('ff01::1'), ('ff01::1', {'subs': {'type': 'multicast', 'scope': 'interface-local'}}))
+
+            info = {'subs': {'type': 'private', 'scope': 'global'}}
             self.eq(t.norm('2001:0db8:0000:0000:0000:ff00:0042:8329'), ('2001:db8::ff00:42:8329', info))
             self.eq(t.norm('2001:0db8:0000:0000:0000:ff00:0042\u200b:8329'), ('2001:db8::ff00:42:8329', info))
             self.raises(s_exc.BadTypeValu, t.norm, 'newp')
