Merge pull request #1457 from p-l-/fix-cert-pk

DB: fix cert pubkey lookups

ivre/db/__init__.py

@@ -1824,7 +1824,7 @@ def searchcert(
             hashval = locals()[f"pk{hashtype}"]
             if hashval is None:
                 continue
-            key = f"pk{hashtype}"
+            key = f"pubkey.{hashtype}"
             if isinstance(hashval, utils.REGEXP_T):
                 values[key] = re.compile(hashval.pattern, hashval.flags | re.I)
                 continue

ivre/db/elastic.py

@@ -1617,7 +1617,7 @@ def searchcert(
             hashval = locals()[f"pk{hashtype}"]
             if hashval is None:
                 continue
-            key = f"ports.scripts.ssl-cert.pk{hashtype}"
+            key = f"ports.scripts.ssl-cert.pubkey.{hashtype}"
             if isinstance(hashval, utils.REGEXP_T):
                 req.append(Q("regexp", **{key: cls._get_pattern(hashval).lower()}))
                 continue

ivre/db/mongo.py

@@ -793,7 +793,7 @@ def _searchcert(
             hashval = locals()[f"pk{hashtype}"]
             if hashval is None:
                 continue
-            key = f"{prefix}pk{hashtype}"
+            key = f"{prefix}pubkey.{hashtype}"
             if isinstance(hashval, utils.REGEXP_T):
                 res[key] = re.compile(hashval.pattern, hashval.flags | re.I)
                 continue

ivre/db/sql/__init__.py

@@ -562,7 +562,7 @@ def _searchcert(
             hashval = locals()[f"pk{hashtype}"]
             if hashval is None:
                 continue
-            key = base.op("->>")("pk{hashtype}")
+            key = base.op("->")("pubkey").op("->>")(hashtype)
             if isinstance(hashval, utils.REGEXP_T):
                 req &= key.op("~*")(hashval.pattern)
                 continue

ivre/db/sql/postgres.py

@@ -728,14 +728,19 @@ def topvalues(
             )
         elif field.startswith("cert."):
             subfield = field[5:]
+            topfld = func.jsonb_array_elements(self.tables.script.data["ssl-cert"])
+            if "." in subfield:
+                first_fields = subfield.split(".")
+                last_field = first_fields.pop()
+                for subfld in first_fields:
+                    topfld = topfld.op("->")(subfld)
+                topfld = topfld.op("->>")(last_field)
+            else:
+                topfld = topfld.op(
+                    "->" if subfield in ["subject", "issuer", "pubkey"] else "->>"
+                )(subfield)
             field = self._topstructure(
-                self.tables.script,
-                [
-                    func.jsonb_array_elements(self.tables.script.data["ssl-cert"],).op(
-                        "->" if subfield in ["subject", "issuer", "pubkey"] else "->>"
-                    )(subfield)
-                ],
-                self.tables.script.name == "ssl-cert",
+                self.tables.script, [topfld], self.tables.script.name == "ssl-cert"
             )
         elif field.startswith("cacert."):
             subfield = field[5:]

ivre/db/tiny.py

@@ -482,7 +482,7 @@ def _searchcert(
             hashval = locals()[f"pk{hashtype}"]
             if hashval is None:
                 continue
-            key = getattr(base, f"pk{hashtype}")
+            key = getattr(base.pubkey, hashtype)
             if isinstance(hashval, utils.REGEXP_T):
                 res &= key.search(hashval.pattern, flags=hashval.flags | re.I)
                 continue

ivre/web/utils.py

@@ -492,7 +492,16 @@ def add_unused(neg, param, value):
                 flt = dbase.flt_and(
                     flt, dbase.searchcert(self_signed=not neg, cacert=cacert)
                 )
-            elif subfield in {"md5", "sha1", "sha256", "subject", "issuer"}:
+            elif subfield in {
+                "md5",
+                "sha1",
+                "sha256",
+                "subject",
+                "issuer",
+                "pkmd5",
+                "pksha1",
+                "pksha256",
+            }:
                 flt = dbase.flt_and(
                     flt,
                     dbase.searchcert(

tests/samples/results

@@ -224,6 +224,9 @@ nmap_top_categories = ['TEST']
 nmap_top_categories_TEST = ['TEST']
 nmap_top_cert_issuer = [{'organizationName': 'CradlePoint Technology', 'organizationalUnitName': 'http://www.cradlepoint.com', 'commonName': 'cp'}]
 nmap_top_cert_md5 = ['ac26955bfd54108acd55486fa62944c3']
+nmap_top_cert_pkmd5 = ['7d938833f3e8edf0c4c3be1fc8612d68']
+nmap_top_cert_pksha1 = ['bb8de26693b134d645c6d81a67b07bbcfbed92d0']
+nmap_top_cert_pksha256 = ['b5b8b2192df85fc14ef7a53c98f8ae826909f0cd1f57cb84445b7be941c4f3dd']
 nmap_top_cert_sha1 = ['397514bb841f893fdd9c3442cc69758288272ff0']
 nmap_top_cert_sha256 = ['cdf555315a8ba0139e721d5b17cbee6599c42b52e4b8eb52f9237825b1e74f64']
 nmap_top_cert_subject = [{'organizationName': 'CradlePoint Technology', 'organizationalUnitName': 'http://www.cradlepoint.com', 'commonName': 'cp'}]
@@ -519,6 +522,9 @@ view_top_categories = ['TEST']
 view_top_categories_PASSIVE = ['PASSIVE']
 view_top_cert_issuer = [{'organizationName': 'CradlePoint Technology', 'organizationalUnitName': 'http://www.cradlepoint.com', 'commonName': 'cp'}]
 view_top_cert_md5 = ['ac26955bfd54108acd55486fa62944c3']
+view_top_cert_pkmd5 = ['7d938833f3e8edf0c4c3be1fc8612d68']
+view_top_cert_pksha1 = ['bb8de26693b134d645c6d81a67b07bbcfbed92d0']
+view_top_cert_pksha256 = ['b5b8b2192df85fc14ef7a53c98f8ae826909f0cd1f57cb84445b7be941c4f3dd']
 view_top_cert_sha1 = ['397514bb841f893fdd9c3442cc69758288272ff0']
 view_top_cert_sha256 = ['cdf555315a8ba0139e721d5b17cbee6599c42b52e4b8eb52f9237825b1e74f64']
 view_top_cert_subject = [{'organizationName': 'CradlePoint Technology', 'organizationalUnitName': 'http://www.cradlepoint.com', 'commonName': 'cp'}]

tests/tests.py

@@ -1697,6 +1697,26 @@ def _has_fingerprint(host):
                         if found:
                             break
                     self.assertTrue(found)
+            self.check_nmap_top_value(
+                "nmap_top_cert_pk%s" % hashtype, "cert.pubkey.%s" % hashtype
+            )
+            for val in self._sort_top_values(
+                ivre.db.db.nmap.topvalues("cert.pubkey.%s" % hashtype)
+            ):
+                for host in ivre.db.db.nmap.get(
+                    ivre.db.db.nmap.searchcert(**{f"pk{hashtype}": val})
+                ):
+                    found = False
+                    for port in host["ports"]:
+                        for script in port.get("scripts", []):
+                            if script["id"] == "ssl-cert":
+                                for cert in script.get("ssl-cert", []):
+                                    if cert.get("pubkey", {}).get(hashtype) == val:
+                                        found = True
+                                        break
+                        if found:
+                            break
+                    self.assertTrue(found)
         self._check_top_value_cli("nmap_top_filename", "file", command="scancli")
         self._check_top_value_cli(
             "nmap_top_filename", "file.filename", command="scancli"
@@ -5391,6 +5411,26 @@ def test_50_view(self):
                         if found:
                             break
                     self.assertTrue(found)
+            self.check_view_top_value(
+                "view_top_cert_pk%s" % hashtype, "cert.pubkey.%s" % hashtype
+            )
+            for val in self._sort_top_values(
+                ivre.db.db.view.topvalues("cert.pubkey.%s" % hashtype)
+            ):
+                for host in ivre.db.db.view.get(
+                    ivre.db.db.view.searchcert(**{f"pk{hashtype}": val})
+                ):
+                    found = False
+                    for port in host["ports"]:
+                        for script in port.get("scripts", []):
+                            if script["id"] == "ssl-cert":
+                                for cert in script.get("ssl-cert", []):
+                                    if cert.get("pubkey", {}).get(hashtype) == val:
+                                        found = True
+                                        break
+                        if found:
+                            break
+                    self.assertTrue(found)
         self.check_view_top_value("view_top_filename", "file")
         self.check_view_top_value("view_top_filename", "file.filename")
         self.check_view_top_value("view_top_anonftp_filename", "file:ftp-anon")

web/static/ivre/content.js

@@ -877,6 +877,18 @@ var HELP_TOPVALUES = {
 	    "content": "cert.sha256",
 	    "title": "<b>(!)</b>cert.sha256"
 	},
+	"cert.pubkey.md5": {
+	    "content": "cert.pubkey.md5",
+	    "title": "<b>(!)</b>cert.pubkey.md5"
+	},
+	"cert.pubkey.sha1": {
+	    "content": "cert.pubkey.sha1",
+	    "title": "<b>(!)</b>cert.pubkey.sha1"
+	},
+	"cert.pubkey.sha256": {
+	    "content": "cert.pubkey.sha256",
+	    "title": "<b>(!)</b>cert.pubkey.sha256"
+	},
 	"sshkey.type": {
 	    "title": "<b>(!)</b>sshkey.type",
 	    "content": "Most common SSH host key types."

web/static/ivre/graph.js

@@ -178,6 +178,11 @@ var GraphTopValues = (function(_super) {
 		    return 'setparam(FILTER, "' + field + '", "' + x + '")';
 		};
 	    }
+	    else if(['cert.pubkey.md5', 'cert.pubkey.sha1', 'cert.pubkey.sha256'].indexOf(field) !== -1) {
+		preparefilter = function(x) {
+		    return 'setparam(FILTER, "cert.pk' + field.substr(12) + '", "' + x + '")';
+		};
+	    }
 	    else if(field === 'asnum') {
 		preparefilter = function(x) {
 		    return 'setparam(FILTER, "asnum", "' + x + '", true);';