diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/sec_certs/constants.py | 7 | ||||
| -rw-r--r-- | src/sec_certs/dataset/cc.py | 23 | ||||
| -rw-r--r-- | src/sec_certs/sample/cc.py | 58 | ||||
| -rw-r--r-- | src/sec_certs/sample/cc_scheme.py | 216 | ||||
| -rw-r--r-- | src/sec_certs/utils/sanitization.py | 9 |
5 files changed, 153 insertions, 160 deletions
diff --git a/src/sec_certs/constants.py b/src/sec_certs/constants.py index e9b6d882..69f0c6cd 100644 --- a/src/sec_certs/constants.py +++ b/src/sec_certs/constants.py @@ -145,7 +145,6 @@ CC_SWEDEN_INEVAL_URL = CC_SWEDEN_BASE_URL + "/verksamhet/ovrig-verksamhet/csec/p CC_SWEDEN_ARCHIVED_URL = CC_SWEDEN_BASE_URL + "/verksamhet/ovrig-verksamhet/csec/arkiverade-certifikat-aldre-an-5-ar/" CC_TURKEY_ARCHIVED_URL = "https://statik.tse.org.tr/upload/tr/dosya/icerikyonetimi/3300/03112021143434-2.pdf" CC_USA_BASE_URL = "https://www.niap-ccevs.org" -CC_USA_PRODUCT_URL = CC_USA_BASE_URL + "/Product/" -CC_USA_CERTIFIED_URL = CC_USA_BASE_URL + "/Product/PCL.cfm" -CC_USA_INEVAL_URL = CC_USA_BASE_URL + "/Product/PINE.cfm" -CC_USA_ARCHIVED_URL = CC_USA_BASE_URL + "/Product/Archived.cfm" +CC_USA_PRODUCTS_URL = CC_USA_BASE_URL + "/api/project/product/pcl_products/" +CC_USA_FILES_URL = CC_USA_BASE_URL + "/api/file/get_pcl_files/" +CC_USA_GETFILE_URL = CC_USA_BASE_URL + "/api/file/get_public_file/" diff --git a/src/sec_certs/dataset/cc.py b/src/sec_certs/dataset/cc.py index 8e6ecebb..ebffd912 100644 --- a/src/sec_certs/dataset/cc.py +++ b/src/sec_certs/dataset/cc.py @@ -34,7 +34,7 @@ from sec_certs.sample.cc_maintenance_update import CCMaintenanceUpdate from sec_certs.sample.cc_scheme import EntryType from sec_certs.sample.protection_profile import ProtectionProfile from sec_certs.serialization.json import ComplexSerializableType, serialize -from sec_certs.utils import helpers +from sec_certs.utils import helpers, sanitization from sec_certs.utils import parallel_processing as cert_processing from sec_certs.utils.profiling import staged @@ -368,7 +368,14 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable return CCDataset.BASE_URL + relative_path def _get_primary_key_str(row: Tag): - return row["category"] + row["cert_name"] + row["report_link"] + return "|".join( + [ + row["category"], + row["cert_name"], + sanitization.sanitize_link_fname(row["report_link"]) or "None", + sanitization.sanitize_link_fname(row["st_link"]) or "None", + ] + ) cert_status = "active" if "active" in str(file) else "archived" @@ -408,11 +415,15 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable df_base = df.loc[~df.is_maintenance].copy() df_main = df.loc[df.is_maintenance].copy() - df_base.report_link = df_base.report_link.map(map_ip_to_hostname) - df_base.st_link = df_base.st_link.map(map_ip_to_hostname) + df_base.report_link = df_base.report_link.map(map_ip_to_hostname).map(sanitization.sanitize_link) + df_base.st_link = df_base.st_link.map(map_ip_to_hostname).map(sanitization.sanitize_link) - df_main.maintenance_report_link = df_main.maintenance_report_link.map(map_ip_to_hostname) - df_main.maintenance_st_link = df_main.maintenance_st_link.map(map_ip_to_hostname) + df_main.maintenance_report_link = df_main.maintenance_report_link.map(map_ip_to_hostname).map( + sanitization.sanitize_link + ) + df_main.maintenance_st_link = df_main.maintenance_st_link.map(map_ip_to_hostname).map( + sanitization.sanitize_link + ) n_all = len(df_base) n_deduplicated = len(df_base.drop_duplicates(subset=["dgst"])) diff --git a/src/sec_certs/sample/cc.py b/src/sec_certs/sample/cc.py index 9f8ba2bd..5354ab3b 100644 --- a/src/sec_certs/sample/cc.py +++ b/src/sec_certs/sample/cc.py @@ -15,7 +15,6 @@ from bs4 import Tag import sec_certs.utils.extract import sec_certs.utils.pdf -import sec_certs.utils.sanitization from sec_certs import constants from sec_certs.cert_rules import SARS_IMPLIED_FROM_EAL, cc_rules, rules, security_level_csv_scan from sec_certs.configuration import config @@ -27,7 +26,7 @@ from sec_certs.sample.protection_profile import ProtectionProfile from sec_certs.sample.sar import SAR from sec_certs.serialization.json import ComplexSerializableType from sec_certs.serialization.pandas import PandasSerializableType -from sec_certs.utils import helpers +from sec_certs.utils import helpers, sanitization from sec_certs.utils.extract import normalize_match_string, scheme_frontpage_functions @@ -57,16 +56,10 @@ class CCCertificate( maintenance_st_link: str | None def __post_init__(self): - super().__setattr__( - "maintenance_report_link", sec_certs.utils.sanitization.sanitize_cc_link(self.maintenance_report_link) - ) - super().__setattr__( - "maintenance_st_link", sec_certs.utils.sanitization.sanitize_cc_link(self.maintenance_st_link) - ) - super().__setattr__( - "maintenance_title", sec_certs.utils.sanitization.sanitize_string(self.maintenance_title) - ) - super().__setattr__("maintenance_date", sec_certs.utils.sanitization.sanitize_date(self.maintenance_date)) + super().__setattr__("maintenance_report_link", sanitization.sanitize_link(self.maintenance_report_link)) + super().__setattr__("maintenance_st_link", sanitization.sanitize_link(self.maintenance_st_link)) + super().__setattr__("maintenance_title", sanitization.sanitize_string(self.maintenance_title)) + super().__setattr__("maintenance_date", sanitization.sanitize_date(self.maintenance_date)) @classmethod def from_dict(cls, dct: dict) -> CCCertificate.MaintenanceReport: @@ -420,20 +413,20 @@ class CCCertificate( self.status = status self.category = category - self.name = sec_certs.utils.sanitization.sanitize_string(name) + self.name = sanitization.sanitize_string(name) self.manufacturer = None if manufacturer: - self.manufacturer = sec_certs.utils.sanitization.sanitize_string(manufacturer) + self.manufacturer = sanitization.sanitize_string(manufacturer) self.scheme = scheme - self.security_level = sec_certs.utils.sanitization.sanitize_security_levels(security_level) - self.not_valid_before = sec_certs.utils.sanitization.sanitize_date(not_valid_before) - self.not_valid_after = sec_certs.utils.sanitization.sanitize_date(not_valid_after) - self.report_link = sec_certs.utils.sanitization.sanitize_cc_link(report_link) - self.st_link = sec_certs.utils.sanitization.sanitize_cc_link(st_link) - self.cert_link = sec_certs.utils.sanitization.sanitize_cc_link(cert_link) - self.manufacturer_web = sec_certs.utils.sanitization.sanitize_link(manufacturer_web) + self.security_level = sanitization.sanitize_security_levels(security_level) + self.not_valid_before = sanitization.sanitize_date(not_valid_before) + self.not_valid_after = sanitization.sanitize_date(not_valid_after) + self.report_link = sanitization.sanitize_link(report_link) + self.st_link = sanitization.sanitize_link(st_link) + self.cert_link = sanitization.sanitize_link(cert_link) + self.manufacturer_web = sanitization.sanitize_link(manufacturer_web) self.protection_profiles = protection_profiles self.maintenance_updates = maintenance_updates self.state = state if state else self.InternalState() @@ -445,6 +438,29 @@ class CCCertificate( """ Computes the primary key of the sample using first 16 bytes of SHA-256 digest """ + if not (self.name is not None and self.category is not None): + raise RuntimeError("Certificate digest can't be computed, because information is missing.") + return helpers.get_first_16_bytes_sha256( + "|".join( + [ + self.category, + self.name, + sanitization.sanitize_link_fname(self.report_link) or "None", + sanitization.sanitize_link_fname(self.st_link) or "None", + ] + ) + ) + + @property + def old_dgst(self) -> str: + if not (self.name is not None and self.report_link is not None and self.category is not None): + raise RuntimeError("Certificate digest can't be computed, because information is missing.") + return helpers.get_first_16_bytes_sha256( + self.category + self.name + sanitization.sanitize_cc_link(self.report_link) # type: ignore + ) + + @property + def older_dgst(self) -> str: if not (self.name is not None and self.report_link is not None and self.category is not None): raise RuntimeError("Certificate digest can't be computed, because information is missing.") return helpers.get_first_16_bytes_sha256(self.category + self.name + self.report_link) diff --git a/src/sec_certs/sample/cc_scheme.py b/src/sec_certs/sample/cc_scheme.py index 1af70e76..2785551d 100644 --- a/src/sec_certs/sample/cc_scheme.py +++ b/src/sec_certs/sample/cc_scheme.py @@ -18,6 +18,7 @@ from urllib.parse import urljoin import requests import tabula from bs4 import BeautifulSoup, NavigableString, Tag +from dateutil.parser import isoparse from requests import Response from urllib3.connectionpool import InsecureRequestWarning @@ -65,12 +66,13 @@ __all__ = [ ] -def _get(url: str, session, **kwargs) -> Response: +def _getq(url: str, params, session=None, **kwargs) -> Response: with warnings.catch_warnings(): warnings.simplefilter("ignore", category=InsecureRequestWarning) conn = session if session else requests resp = conn.get( url, + params=params, headers={"User-Agent": "sec-certs.org"}, verify=False, **kwargs, @@ -80,6 +82,10 @@ def _get(url: str, session, **kwargs) -> Response: return resp +def _get(url: str, session=None, **kwargs) -> Response: + return _getq(url, None, session, **kwargs) + + def _get_page(url: str, session=None) -> BeautifulSoup: return BeautifulSoup(_get(url, session).content, "html5lib") @@ -1306,7 +1312,7 @@ def get_spain_certified() -> list[dict[str, Any]]: "product_link": urljoin(constants.CC_SPAIN_BASE_URL, tds[0].find("a")["href"]), "category": sns(tds[1].text), "manufacturer": sns(tds[2].text), - "certification_date": sns(tds[3].find("td", class_="djc_value").text), + "certification_date": sns(tds[3].text), } results.append(cert) return results @@ -1444,6 +1450,78 @@ def get_turkey_certified() -> list[dict[str, Any]]: return results +def _get_usa(args, enhanced: bool, artifacts: bool): # noqa: C901 + # TODO: There is more information in the API (like about PPs, etc.) + def map_cert(cert, files=None): # noqa: C901 + result = { + "product": cert["product_name"], + "id": f"CCEVS-VR-VID{cert['product_id']}", + "url": constants.CC_USA_BASE_URL + f"/product/{cert['product_id']}", + "certification_date": cert["certification_date"], + "expiration_date": cert["sunset_date"], + "category": cert["tech_type"], + "vendor": cert["vendor_id_name"], + "evaluation_facility": cert["assigned_lab_name"], + "scheme": cert["submitting_country_id_code"], + } + if files: + for file in files["eval_files"]: + if file["file_label"] == "Validation Report": + dt = isoparse(file["uploaded_on"]) + result["id"] += f"-{dt.year}" + result["report_link"] = constants.CC_USA_GETFILE_URL + f"?file_id={file['file_id']}" + if artifacts: + result["report_hash"] = _get_hash(result["report_link"]).hex() + elif file["file_label"] == "CC Certificate": + result["cert_link"] = constants.CC_USA_GETFILE_URL + f"?file_id={file['file_id']}" + if artifacts: + result["cert_hash"] = _get_hash(result["cert_link"]).hex() + elif file["file_label"] == "Security Target": + result["target_link"] = constants.CC_USA_GETFILE_URL + f"?file_id={file['file_id']}" + if artifacts: + result["target_hash"] = _get_hash(result["target_link"]).hex() + elif file["file_label"] == "Assurance Activity Report (AAR)": + result["aar_link"] = constants.CC_USA_GETFILE_URL + f"?file_id={file['file_id']}" + if artifacts: + result["aar_hash"] = _get_hash(result["aar_link"]).hex() + elif file["file_label"] == "Administrative Guide (AGD)": + result["agd_link"] = constants.CC_USA_GETFILE_URL + f"?file_id={file['file_id']}" + if artifacts: + result["agd_hash"] = _get_hash(result["agd_link"]).hex() + + return result + + session = requests.session() + results = [] + offset = 0 + got = 0 + while True: + resp = _getq( + constants.CC_USA_PRODUCTS_URL, + {"limit": 100, "offset": offset, **args}, + session, + ) + json = resp.json() + count = json["count"] + for cert in json["results"]["products"]: + got += 1 + if "from_cc_portal" in cert: + continue + files = None + if enhanced: + resp = _getq( + constants.CC_USA_FILES_URL, + {"product_id": cert["product_id"]}, + session, + ) + files = resp.json() + results.append(map_cert(cert, files)) + offset += 100 + if got >= count: + break + return results + + def get_usa_certified( # noqa: C901 enhanced: bool = True, artifacts: bool = False ) -> list[dict[str, Any]]: @@ -1454,83 +1532,11 @@ def get_usa_certified( # noqa: C901 :param artifacts: Whether to download and compute artifact hashes (way slower, even more data). :return: The entries. """ - # TODO: Information could be expanded by following the cc_claims (has links to protection profiles). - soup = _get_page(constants.CC_USA_CERTIFIED_URL) - tbody = soup.find("table", class_="tablesorter").find("tbody") - results = [] - for tr in tbody.find_all("tr"): - tds = tr.find_all("td") - vendor_span = tds[0].find("span", class_="b u") - product_link = tds[0].find("a") - scheme_img = tds[6].find("img") - # Only return the US certifications. - if scheme_img["title"] != "USA": - continue - cert: dict[str, Any] = { - "product": sns(product_link.text), - "vendor": sns(vendor_span.text), - "product_link": urljoin(constants.CC_USA_PRODUCT_URL, product_link["href"]), - "id": sns(tds[1].text), - "cc_claim": sns(tds[2].text), - "cert_lab": sns(tds[3].text), - "certification_date": sns(tds[4].text), - "assurance_maintenance_date": sns(tds[5].text), - } - if enhanced: - e: dict[str, Any] = {} - if not cert["product_link"]: - continue - cert_page = _get_page(cert["product_link"]) - details = cert_page.find("div", class_="txt2 lma") - for span in details.find_all("span"): - title = sns(span.text) - if not title: - continue - sibling = span.next_sibling - value = sns(sibling.text) - if "Certificate Date" in title: - e["certification_date"] = value - elif "Product Type" in title: - e["product_type"] = value - elif "Conformance Claim" in title: - e["cc_claim"] = value - elif "Validation Report Number" in title: - e["cert_id"] = value - elif "PP Identifier" in title: - e["protection_profile"] = sns(span.find_next_sibling("a").text) - elif "CC Testing Lab" in title: - e["evaluation_facility"] = sns(span.find_next_sibling("a").text) - links = cert_page.find_all("a", class_="pseudobtn1") - for link in links: - name = sns(link.text) - href = urljoin(constants.CC_USA_BASE_URL, sns(link["href"])) - if not name: - continue - if "CC Certificate" in name: - e["cert_link"] = href - if artifacts: - e["cert_hash"] = _get_hash(href).hex() - elif "Security Target" in name: - e["target_link"] = href - if artifacts: - e["target_hash"] = _get_hash(href).hex() - elif "Validation Report" in name: - e["report_link"] = href - if artifacts: - e["report_hash"] = _get_hash(href).hex() - elif "Assurance Activity" in name: - e["assurance_activity_link"] = href - if artifacts: - e["assurance_activity_hash"] = _get_hash(href).hex() - elif "Administrative Guide" in name: - guides = e.setdefault("administrative_guides", []) - guide = {"link": href} - guides.append(guide) - if artifacts: - guide["hash"] = _get_hash(href).hex() - cert["enhanced"] = e - results.append(cert) - return results + return _get_usa( + {"certification_status": "Certified", "publish_status": "Published"}, + enhanced, + artifacts, + ) def get_usa_in_evaluation() -> list[dict[str, Any]]: @@ -1539,29 +1545,7 @@ def get_usa_in_evaluation() -> list[dict[str, Any]]: :return: The entries. """ - # TODO: Information could be expanded by following the cc_claims (has links to protection profiles). - soup = _get_page(constants.CC_USA_INEVAL_URL) - tbody = soup.find("table", class_="tablesorter").find("tbody") - results = [] - for tr in tbody.find_all("tr"): - tds = tr.find_all("td") - vendor_span = tds[0].find("span", class_="b u") - product_name = None - for child in tds[0].children: - if isinstance(child, NavigableString): - product_name = sns(child) - break - cert = { - "vendor": sns(vendor_span.text), - "id": sns(tds[1].text), - "cc_claim": sns(tds[2].text), - "cert_lab": sns(tds[3].text), - "kickoff_date": sns(tds[4].text), - } - if product_name: - cert["product"] = product_name - results.append(cert) - return results + return _get_usa({"status": "In Progress", "publish_status": "Published"}, False, False) def get_usa_archived() -> list[dict[str, Any]]: @@ -1570,33 +1554,7 @@ def get_usa_archived() -> list[dict[str, Any]]: :return: The entries. """ - # TODO: Information could be expanded by following the cc_claims (has links to protection profiles). - soup = _get_page(constants.CC_USA_ARCHIVED_URL) - tbody = soup.find("table", class_="tablesorter").find("tbody") - results = [] - for tr in tbody.find_all("tr"): - tds = tr.find_all("td") - scheme_img = tds[5].find("img") - # Only return the US certifications. - if scheme_img["title"] != "USA": - continue - vendor_span = tds[0].find("span", class_="b u") - product_name = None - for child in tds[0].children: - if isinstance(child, NavigableString): - product_name = sns(child) - break - cert = { - "vendor": sns(vendor_span.text), - "id": sns(tds[1].text), - "cc_claim": sns(tds[2].text), - "cert_lab": sns(tds[3].text), - "certification_date": sns(tds[4].text), - } - if product_name: - cert["product"] = product_name - results.append(cert) - return results + return _get_usa({"status": "Archived", "publish_status": "Published"}, False, False) class EntryType(Enum): diff --git a/src/sec_certs/utils/sanitization.py b/src/sec_certs/utils/sanitization.py index fd477746..4a7326eb 100644 --- a/src/sec_certs/utils/sanitization.py +++ b/src/sec_certs/utils/sanitization.py @@ -3,6 +3,8 @@ from __future__ import annotations import html import logging from datetime import date +from pathlib import Path +from urllib.parse import urlparse import numpy as np import pandas as pd @@ -23,6 +25,13 @@ def sanitize_link(record: str | None) -> str | None: return record.replace(":443", "").replace(" ", "%20").replace("http://", "https://") +def sanitize_link_fname(record: str | None) -> str | None: + if not record: + return None + parsed = urlparse(record) + return Path(parsed.path).name + + def sanitize_cc_link(record: str | None) -> str | None: record = sanitize_link(record) if not record: |