Merge pull request #189 from vertexproject/visi-ingest

new infotech model and av / cve update

synapse/__init__.py

@@ -32,9 +32,9 @@
 s_modules.load('synapse.models.crypto')
 s_modules.load('synapse.models.geopol')
 s_modules.load('synapse.models.person')
+s_modules.load('synapse.models.infotech')
 s_modules.load('synapse.models.temporal')
 s_modules.load('synapse.models.geospace')
-s_modules.load('synapse.models.av')
 
 mods = os.getenv('SYN_MODULES')
 if mods:

synapse/models/infotech.py

@@ -0,0 +1,36 @@
+def getDataModel():
+    return {
+        'prefix':'it',
+        'version':201703301552,
+
+        'types': (
+            ('it:host',         {'subof':'guid','doc':'A GUID for a host/system'}),
+            ('it:sec:cve',      {'subof':'str', 'regex':'^CVE-[0-9]{4}-[0-9]{4,6}$','doc':'A CVE entry from Mitre'}),
+
+            ('it:av:sig',       {'subof':'sepr', 'sep':'/', 'fields':'org,ou:alias|sig,str:lwr', 'doc':'An antivirus signature' }),
+            ('it:av:filehit',   {'subof':'sepr', 'sep':'/', 'fields':'file,file:bytes|sig,it:av:sig', 'doc':'An antivirus hit' }),
+        ),
+
+        'forms': (
+
+            ('it:host', {'ptype':'it:host'},[
+                #FIXME we probably eventually need a bunch of stuff here...
+            ]),
+
+            ('it:sec:cve', {'ptype':'it:sec:cve'},[
+                ('desc',{'ptype':'str'}),
+            ]),
+
+            ('it:av:sig', {'ptype':'it:av:sig'},[
+                ('sig',{'ptype':'str:lwr'}),
+                ('org',{'ptype':'ou:alias'}),
+                ('desc',{'ptype':'str'}),
+                ('url',{'ptype':'inet:url'}),
+            ]),
+
+            ('it:av:filehit', {'ptype':'it:av:filehit'},[
+                ('file',{'ptype':'file:bytes'}),
+                ('sig',{'ptype':'it:av:sig'}),
+            ])
+        ),
+    }

synapse/models/orgs.py

@@ -5,9 +5,7 @@ def getDataModel():
 
         'types':(
             ('ou:org',{'subof':'guid','doc':'A GUID for a human organization such as a company or military unit'}),
-            ('ou:host',{'subof':'guid','doc':'A GUID for a host within an organization'}),
             ('ou:user',{'subof':'sepr','sep':'/','fields':'org,ou:org|user,inet:user','doc':'A user name within an organization'}),
-
             ('ou:alias',{'subof':'str','lower':1,'regex':'^[0-9a-z]+$','doc':'An alias for the org GUID','ex':'vertexproj'}),
         ),
 

synapse/tests/common.py

@@ -129,6 +129,9 @@ def noprop(self, info, prop):
         valu = info.get(prop,novalu)
         self.eq(valu,novalu)
 
+    def raises(self, *args, **kwargs):
+        return self.assertRaises(*args,**kwargs)
+
     def sorteq(self, x, y):
         return self.eq( sorted(x), sorted(y) )
 

synapse/tests/test_model_infotech.py

@@ -0,0 +1,40 @@
+
+from synapse.tests.common import *
+
+class InfoTechTest(SynTest):
+
+    def test_model_infotech_host(self):
+        with s_cortex.openurl('ram:///') as core:
+            core.setConfOpt('enforce',1)
+            node = core.formTufoByProp('it:host',guid())
+            self.nn(node)
+            self.nn(node[1].get('it:host'))
+
+    def test_model_infotech_cve(self):
+        with s_cortex.openurl('ram:///') as core:
+            core.setConfOpt('enforce',1)
+            node = core.formTufoByProp('it:sec:cve','CVE-2013-9999', desc='This is a description')
+            self.nn(node)
+            self.eq( node[1].get('it:sec:cve'), 'CVE-2013-9999')
+            self.eq( node[1].get('it:sec:cve:desc'), 'This is a description' )
+            self.raises( BadTypeValu, core.formTufoByProp, 'it:sec:cve', 'dERP' )
+
+    def test_model_infotech_av(self):
+        with s_cortex.openurl('ram:///') as core:
+            core.setConfOpt('enforce',1)
+            bytesguid = '1234567890ABCDEFFEDCBA0987654321'
+            orgname = 'Foo'
+            signame = 'Bar.BAZ.faZ'
+            valu = (bytesguid, (orgname, signame))
+
+            tufo = core.formTufoByFrob('it:av:filehit', valu)
+            self.eq(tufo[1].get('it:av:filehit:sig'), 'foo/bar.baz.faz')
+            self.eq(tufo[1].get('it:av:filehit:file'), '1234567890abcdeffedcba0987654321')
+
+            tufo = core.getTufoByProp('it:av:sig', 'foo/bar.baz.faz')
+            self.eq(tufo[1].get('it:av:sig'), 'foo/bar.baz.faz')
+            self.eq(tufo[1].get('it:av:sig:org'), 'foo')
+            self.eq(tufo[1].get('it:av:sig:sig'), 'bar.baz.faz')
+
+            tufo = core.getTufoByProp('ou:alias', 'foo')
+            self.eq(tufo, None) # ou:alias will not be automatically formed at this time

synapse/tests/test_types.py

@@ -595,20 +595,3 @@ def cast(x):
         self.eq( tlib.getTypeCast('str:lwr','HeHe'), 'hehe' )
         self.eq( tlib.getTypeCast('toupper','HeHe'), 'HEHE' )
         self.eq( tlib.getTypeCast('make:guid','visi'), '1b2e93225959e3722efed95e1731b764' )
-
-    def test_type_av_types(self):
-        tlib = s_types.TypeLib()
-
-        sig = 'tech:av:sig'
-        hit = 'tech:av:filehit'
-        orgname = 'fOo'
-        bytesguid = 'FEDCBA0987654321FEDCBA0987654321'
-        signame = 'BaR.BAZ.faz'
-
-        self.eq(tlib.getTypeNorm(sig, '%s/%s'%(orgname, signame))[0], '%s/%s'%(orgname.lower(), signame.lower()))
-        self.eq(tlib.getTypeFrob(sig, '%s/%s'%(orgname, signame))[0], '%s/%s'%(orgname.lower(), signame.lower()))
-        self.eq(tlib.getTypeParse(sig, '%s/%s'%(orgname, signame))[0], '%s/%s'%(orgname.lower(), signame.lower()))
-
-        self.eq(tlib.getTypeNorm(hit, '%s/%s/%s'%(bytesguid, orgname, signame))[0], '%s/%s/%s'%(bytesguid.lower(), orgname.lower(), signame.lower()))
-        self.eq(tlib.getTypeFrob(hit, '%s/%s/%s'%(bytesguid, orgname, signame))[0], '%s/%s/%s'%(bytesguid.lower(), orgname.lower(), signame.lower()))
-        self.eq(tlib.getTypeParse(hit, '%s/%s/%s'%(bytesguid, orgname, signame))[0], '%s/%s/%s'%(bytesguid.lower(), orgname.lower(), signame.lower()))