diff options
| author | J08nY | 2024-11-08 00:17:39 +0100 |
|---|---|---|
| committer | J08nY | 2024-11-08 00:17:39 +0100 |
| commit | 17ae4dbfe97d8f7ef41ea272325d85de9f731af1 (patch) | |
| tree | f3059a02b9327c4c159d20af646cf27f5a9b7965 /src | |
| parent | 11a7052fdd1bf0b01d9dad01e5d6628efc2c9430 (diff) | |
| download | sec-certs-17ae4dbfe97d8f7ef41ea272325d85de9f731af1.tar.gz sec-certs-17ae4dbfe97d8f7ef41ea272325d85de9f731af1.tar.zst sec-certs-17ae4dbfe97d8f7ef41ea272325d85de9f731af1.zip | |
Improve CC scheme extraction and matching.
This significantly improves the CC scheme extraction by:
- Fixing the extraction of several schemes that were mixing
certified and archived entries by accident.
- Improving the extraction of cert_ids from scheme sites.
- Improving the matching heuristic to consider more attributes
that are usually present in the site data.
Also adds an evaluation notebook to see how this performs.
Diffstat (limited to 'src')
| -rw-r--r-- | src/sec_certs/configuration.py | 2 | ||||
| -rw-r--r-- | src/sec_certs/constants.py | 4 | ||||
| -rw-r--r-- | src/sec_certs/dataset/cc.py | 4 | ||||
| -rw-r--r-- | src/sec_certs/model/cc_matching.py | 84 | ||||
| -rw-r--r-- | src/sec_certs/model/matching.py | 8 | ||||
| -rw-r--r-- | src/sec_certs/rules.yaml | 2 | ||||
| -rw-r--r-- | src/sec_certs/sample/cc_certificate_id.py | 18 | ||||
| -rw-r--r-- | src/sec_certs/sample/cc_scheme.py | 320 | ||||
| -rw-r--r-- | src/sec_certs/sample/certificate.py | 2 | ||||
| -rw-r--r-- | src/sec_certs/utils/extract.py | 23 | ||||
| -rw-r--r-- | src/sec_certs/utils/helpers.py | 16 | ||||
| -rw-r--r-- | src/sec_certs/utils/sanitization.py | 5 | ||||
| -rw-r--r-- | src/sec_certs/utils/tqdm.py | 2 |
13 files changed, 324 insertions, 166 deletions
diff --git a/src/sec_certs/configuration.py b/src/sec_certs/configuration.py index 0305bdea..09e24539 100644 --- a/src/sec_certs/configuration.py +++ b/src/sec_certs/configuration.py @@ -122,7 +122,7 @@ class Configuration(BaseSettings): description="True if new reference annotator model shall be build, False otherwise.", ) cc_matching_threshold: int = Field( - 90, + 70, description="Level of required similarity before CC scheme entry is considered to match a CC certificate.", ge=0, le=100, diff --git a/src/sec_certs/constants.py b/src/sec_certs/constants.py index 0e4de22b..d134b3fd 100644 --- a/src/sec_certs/constants.py +++ b/src/sec_certs/constants.py @@ -128,7 +128,9 @@ CC_NORWAY_ARCHIVED_URL = CC_NORWAY_BASE_URL + "/certified-products/product-archi CC_KOREA_BASE_URL = "https://itscc.kr" CC_KOREA_EN_URL = CC_KOREA_BASE_URL + "/main/mainEn.do" CC_KOREA_CERTIFIED_URL = CC_KOREA_BASE_URL + "/certprod/listA.do" -CC_KOREA_PRODUCT_URL = CC_KOREA_BASE_URL + "/certprod/view.do?product_id={}&product_class=1" +CC_KOREA_SUSPENDED_URL = CC_KOREA_BASE_URL + "/certprod/listB.do" +CC_KOREA_ARCHIVED_URL = CC_KOREA_BASE_URL + "/certprod/listD.do" +CC_KOREA_PRODUCT_URL = CC_KOREA_BASE_URL + "/certprod/view.do?product_id={}&product_class={}" CC_POLAND_BASE_URL = "https://en.nask.pl" CC_POLAND_CERTIFIED_URL = CC_POLAND_BASE_URL + "/eng/activities/certification/list-of-certificates" CC_POLAND_INEVAL_URL = CC_POLAND_BASE_URL + "/eng/activities/certification/ongoing-certifications" diff --git a/src/sec_certs/dataset/cc.py b/src/sec_certs/dataset/cc.py index ba2ffb58..d5417a23 100644 --- a/src/sec_certs/dataset/cc.py +++ b/src/sec_certs/dataset/cc.py @@ -870,12 +870,12 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable for scheme in self.auxiliary_datasets.scheme_dset: if certified := scheme.lists.get(EntryType.Certified): certs = [cert for cert in self if cert.status == "active"] - matches = CCSchemeMatcher.match_all(certified, scheme.country, certs) + matches, scores = CCSchemeMatcher.match_all(certified, scheme.country, certs) for dgst, match in matches.items(): self[dgst].heuristics.scheme_data = match if archived := scheme.lists.get(EntryType.Archived): certs = [cert for cert in self if cert.status == "archived"] - matches = CCSchemeMatcher.match_all(archived, scheme.country, certs) + matches, scores = CCSchemeMatcher.match_all(archived, scheme.country, certs) for dgst, match in matches.items(): self[dgst].heuristics.scheme_data = match diff --git a/src/sec_certs/model/cc_matching.py b/src/sec_certs/model/cc_matching.py index 9930fb66..d30a00dd 100644 --- a/src/sec_certs/model/cc_matching.py +++ b/src/sec_certs/model/cc_matching.py @@ -1,13 +1,17 @@ from __future__ import annotations import contextlib +import re from collections.abc import Iterable, Mapping, Sequence +from operator import itemgetter from typing import Any +from sec_certs.cert_rules import rules from sec_certs.configuration import config from sec_certs.model.matching import AbstractMatcher from sec_certs.sample.cc import CCCertificate -from sec_certs.sample.cc_certificate_id import CertificateId +from sec_certs.sample.cc_certificate_id import CertificateId, schemes +from sec_certs.utils.sanitization import sanitize_link_fname from sec_certs.utils.strings import fully_sanitize_string CATEGORIES = { @@ -40,21 +44,23 @@ class CCSchemeMatcher(AbstractMatcher[CCCertificate]): self.scheme = scheme self._prepare() - def _get_from_entry(self, *keys: str) -> str | None: - for key in keys: - if val := self.entry.get(key): - return val + def _get_from_entry(self, *keys: str) -> Any | None: + # Prefer enhanced over base if e := self.entry.get("enhanced"): for key in keys: if val := e.get(key): return val + for key in keys: + if val := self.entry.get(key): + return val return None - def _prepare(self): + def _prepare(self): # noqa: C901 self._canonical_cert_id = None - if cert_id := self._get_from_entry("cert_id", "id"): + self._cert_id = self._get_from_entry("cert_id", "id") + if self._cert_id: with contextlib.suppress(Exception): - self._canonical_cert_id = CertificateId(self.scheme, cert_id).canonical + self._canonical_cert_id = CertificateId(self.scheme, self._cert_id).canonical self._product = None if product_name := self._get_from_entry("product", "title", "name"): @@ -65,11 +71,39 @@ class CCSchemeMatcher(AbstractMatcher[CCCertificate]): self._vendor = fully_sanitize_string(vendor_name) self._category = self._get_from_entry("category") + self._certification_date = self._get_from_entry("certification_date") + self._expiration_date = self._get_from_entry("expiration_date") + self._level = self._get_from_entry("level", "assurance_level") + if self._level: + self._level = self._level.upper().replace("AUGMENTED", "").replace("WITH", "") + + filename_rules = rules["cc_filename_cert_id"][self.scheme] + scheme_meta = schemes[self.scheme] + if filename_rules and self._canonical_cert_id is None: + cert_link = self._get_from_entry("cert_link") + if cert_link: + cert_fname = sanitize_link_fname(cert_link) + for rule in filename_rules: + if match := re.match(rule, cert_fname): + with contextlib.suppress(Exception): + meta = match.groupdict() + self._canonical_cert_id = scheme_meta(meta) + break + + report_link = self._get_from_entry("report_link") + if report_link and self._canonical_cert_id is None: + report_fname = sanitize_link_fname(report_link) + for rule in filename_rules: + if match := re.match(rule, report_fname): + with contextlib.suppress(Exception): + meta = match.groupdict() + self._canonical_cert_id = scheme_meta(meta) + break self._report_hash = self._get_from_entry("report_hash") self._target_hash = self._get_from_entry("target_hash") - def match(self, cert: CCCertificate) -> float: + def match(self, cert: CCCertificate) -> float: # noqa: C901 """ Compute the match of this matcher to the certificate, a float from 0 to 100. @@ -105,21 +139,47 @@ class CCSchemeMatcher(AbstractMatcher[CCCertificate]): return 93 # Fuzzy match at the end with some penalization. + # Weigh the name and vendor more than the id and more than the level and certification date. + # 6, 6, 4, 2, 2 + matches = {} product_rating = self._compute_match(self._product, cert_name) + matches["product"] = (product_rating, 6) vendor_rating = self._compute_match(self._vendor, cert_manufacturer) - return max((0, product_rating * 0.5 + vendor_rating * 0.5 - 2)) + matches["vendor"] = (vendor_rating, 6) + + if self._cert_id is not None and cert.heuristics.cert_id is not None: + id_rating = self._compute_match(self._cert_id, cert.heuristics.cert_id) + matches["id"] = (id_rating, 4) + + if self._certification_date is not None and cert.not_valid_before is not None: + date_rating = 1 + if cert.not_valid_before.year == self._certification_date.year: + date_rating += 33 + if cert.not_valid_before.month == self._certification_date.month: + date_rating += 33 + if cert.not_valid_before.day == self._certification_date.day: + date_rating += 33 + matches["certification_date"] = (date_rating, 2) + + if self._level is not None and cert.security_level: + level_rating = self._compute_match(self._level, ", ".join(cert.security_level)) + matches["level"] = (level_rating, 2) + total_weight = sum(map(itemgetter(1), matches.values())) + return max((0, sum(match[0] * (match[1] / total_weight) for match in matches.values()) - 2)) @classmethod def match_all( cls, entries: list[dict[str, Any]], scheme: str, certificates: Iterable[CCCertificate] - ) -> dict[str, dict[str, Any]]: + ) -> tuple[dict[str, dict[str, Any]], dict[str, float]]: """ Match all entries of a given CC scheme to certificates from the dataset. :param entries: The entries from the scheme, obtained from CCSchemeDataset. :param scheme: The scheme, e.g. "DE". :param certificates: The certificates to match against. - :return: A mapping of certificate digests to entries, without duplicates, not all entries may be present. + :return: Two mappings: + - A mapping of certificate digests to entries, without duplicates, not all entries may be present. + - A mapping of certificate digests to scores that they matched with. """ certs: list[CCCertificate] = list(filter(lambda cert: cert.scheme == scheme, certificates)) matchers: Sequence[CCSchemeMatcher] = [CCSchemeMatcher(entry, scheme) for entry in entries] diff --git a/src/sec_certs/model/matching.py b/src/sec_certs/model/matching.py index bf214e6f..2ba48f64 100644 --- a/src/sec_certs/model/matching.py +++ b/src/sec_certs/model/matching.py @@ -29,7 +29,9 @@ class AbstractMatcher(Generic[CertSubType], ABC): ) @staticmethod - def _match_certs(matchers: Sequence[AbstractMatcher], certs: list[CertSubType], threshold: float): + def _match_certs( + matchers: Sequence[AbstractMatcher], certs: list[CertSubType], threshold: float + ) -> tuple[dict[str, Any], dict[str, float]]: scores: list[tuple[float, int, int]] = [] matched_is: set[int] = set() matched_js: set[int] = set() @@ -39,6 +41,7 @@ class AbstractMatcher(Generic[CertSubType], ABC): triple = (100 - score, i, j) heappush(scores, triple) results = {} + final_scores = {} for triple in (heappop(scores) for _ in range(len(scores))): inv_score, i, j = triple # Do not match already matched entries/certs. @@ -55,4 +58,5 @@ class AbstractMatcher(Generic[CertSubType], ABC): cert = certs[i] entry = matchers[j].entry results[cert.dgst] = entry - return results + final_scores[cert.dgst] = score + return results, final_scores diff --git a/src/sec_certs/rules.yaml b/src/sec_certs/rules.yaml index 16c78aa3..e1e7f1f8 100644 --- a/src/sec_certs/rules.yaml +++ b/src/sec_certs/rules.yaml @@ -16,7 +16,7 @@ cc_cert_id: - "DCSS[Ii]-(?P<year>[0-9]{2,4})/(?P<counter>[0-9]+)([vV](?P<version>[0-9]))?" - "Rapport de certification (?P<year>[0-9]{2,4})/(?P<counter>[0-9]+)([vV](?P<version>[0-9]))?" - "Certification Report (?P<year>[0-9]{2,4})/(?P<counter>[0-9]+)([vV](?P<version>[0-9]))?" - - "ANSS[Ii](?:-CC)?[ -](?P<year>[0-9]{2,4})[/_-](?P<counter>[0-9]+)(?:-(?P<doc>(?:[MSR][0-9]+)))?([vV](?P<version>[0-9]))?" + - "ANSS[Ii](?:-CC)?(?:-(?P<type>PP|SITE))?[ -](?P<year>[0-9]{2,4})[/_-](?P<counter>[0-9]+)(?:-(?P<doc>(?:[MSR][0-9]+)))?([vV](?P<version>[0-9]))?" # Examples: # DCSSI-2009/07 # ANSSI-CC 2001/02-R01 diff --git a/src/sec_certs/sample/cc_certificate_id.py b/src/sec_certs/sample/cc_certificate_id.py index c2b6daa2..74de6b42 100644 --- a/src/sec_certs/sample/cc_certificate_id.py +++ b/src/sec_certs/sample/cc_certificate_id.py @@ -24,7 +24,11 @@ def FR(meta) -> str: counter = meta["counter"] doc = meta.get("doc") version = meta.get("version") - cert_id = f"ANSSI-CC-{year}/{counter}" + type = meta.get("type") + cert_id = "ANSSI-CC-" + if type: + cert_id += f"{type}-" + cert_id += f"{year}/{counter}" if doc: cert_id += f"-{doc}" if version: @@ -183,6 +187,17 @@ def IT(meta) -> str: return cert_id +def PL(meta) -> str: + number = meta["number"] + ac = meta.get("ac") + year = meta["year"] + cert_id = f"{number}/PC1/" + if ac: + cert_id += f"{ac}/" + cert_id += f"{year}" + return cert_id + + # We have rules for some schemes to make canonical cert_ids. schemes = { "FR": FR, @@ -202,6 +217,7 @@ schemes = { "TR": TR, "SG": SG, "IT": IT, + "PL": PL, } diff --git a/src/sec_certs/sample/cc_scheme.py b/src/sec_certs/sample/cc_scheme.py index f02b521b..2cc9fc48 100644 --- a/src/sec_certs/sample/cc_scheme.py +++ b/src/sec_certs/sample/cc_scheme.py @@ -9,7 +9,7 @@ import tempfile import warnings from collections.abc import Callable, Iterable from dataclasses import dataclass -from datetime import datetime +from datetime import date, datetime from enum import Enum from inspect import signature from pathlib import Path @@ -25,6 +25,7 @@ from urllib3.connectionpool import InsecureRequestWarning from sec_certs import constants from sec_certs.serialization.json import ComplexSerializableType +from sec_certs.utils.helpers import parse_date from sec_certs.utils.sanitization import sanitize_navigable_string as sns from sec_certs.utils.tqdm import tqdm @@ -78,21 +79,7 @@ SPOOF_HEADERS = { "Cache-Control": "no-cache", "Dnt": "1", "Pragma": "no-cache", - "Priority": "u=0, i", # 'ICs, Smart Cards and Smart Card-Related Devices and Systems': 1978, - # 'Other Devices and Systems': 1043, - # 'Network and Network-Related Devices and Systems': 835, - # 'Multi-Function Devices': 671, - # 'Boundary Protection Devices and Systems': 253, - # 'Data Protection': 240, - # 'Operating Systems': 237, - # 'Products for Digital Signatures': 179, - # 'Access Control Devices and Systems': 155, - # 'Mobility': 115, - # 'Databases': 103, - # 'Trusted Computing': 78, - # 'Detection Devices and Systems': 77, - # 'Key Management Systems': 59, - # 'Biometric Systems and Devices' + "Priority": "u=0, i", "Sec-Ch-Ua": 'Not?A_Brand";v="99", "Chromium";v="130', "Sec-Ch-Ua-Mobile": "?0", "Sec-Ch-Ua-Platform": '"Linux"', @@ -146,7 +133,7 @@ def get_australia_in_evaluation( # noqa: C901 :param enhanced: Whether to enhance the results by following links (slower, more data). :return: The entries. """ - session = requests.session() + session = requests.Session() soup = _get_page(constants.CC_AUSTRALIA_INEVAL_URL, session=session, spoof=True) header = soup.find("h2", string="Products in evaluation") table = header.find_next_sibling("table") @@ -160,6 +147,9 @@ def get_australia_in_evaluation( # noqa: C901 "product": sns(tds[1].text), "url": urljoin(constants.CC_AUSTRALIA_BASE_URL, tds[1].find("a")["href"]), "level": sns(tds[2].text), + "acceptance_date": parse_date(sns(tds[3].text), languages=["en"]), + "evaluation_facility": sns(tds[4].text), + "task_id": sns(tds[5].text), } if enhanced: e: dict[str, Any] = {} @@ -174,7 +164,7 @@ def get_australia_in_evaluation( # noqa: C901 if "Version:" in h_text: e["version"] = val elif "Product type:" in h_text: - e["product_type"] = val + e["category"] = val elif "Product status:" in h_text: e["product_status"] = val elif "Assurance level:" in h_text: @@ -226,7 +216,7 @@ def get_canada_certified() -> list[dict[str, Any]]: "product": sns(tds[0].text), "vendor": sns(tds[1].text), "level": sns(tds[2].text), - "certification_date": sns(tds[3].text), + "certification_date": parse_date(sns(tds[3].text), "%Y-%m-%d"), } results.append(cert) return results @@ -254,14 +244,14 @@ def get_canada_in_evaluation() -> list[dict[str, Any]]: "product": sns(tds[0].text), "vendor": sns(tds[1].text), "level": sns(tds[2].text), - "cert_lab": sns(tds[3].text), + "evaluation_facility": sns(tds[3].text), } results.append(cert) return results def _get_france(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # noqa: C901 - session = requests.session() + session = requests.Session() challenge_soup = _get_page(constants.CC_ANSSI_BASE_URL, session=session) bln_script = challenge_soup.find("head").find_all("script")[1] bln_match = re.search(r"\"value\":\"([a-zA-Z0-9_-]+)\"", bln_script.string) @@ -296,11 +286,11 @@ def _get_france(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # noqa elif "Développeur" in label: cert["developer"] = value elif "Référence du certificat" in label: - cert["cert_id"] = value if not value or value.startswith("ANSSI") else "ANSSI-CC" + value + cert["cert_id"] = value if not value or value.startswith("ANSSI") else "ANSSI-CC-" + value elif "Niveau" in label: cert["level"] = value elif "Date de fin de validité" in label: - cert["expiration_date"] = value + cert["expiration_date"] = parse_date(value, languages=["fr"]) if enhanced: e: dict[str, Any] = {} cert_page = _get_page(cert["url"], session=session) @@ -311,9 +301,9 @@ def _get_france(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # noqa if "Référence du certificat" in label: e["cert_id"] = value if not value or value.startswith("ANSSI") else "ANSSI-CC-" + value elif "Date de certification" in label: - e["certification_date"] = value + e["certification_date"] = parse_date(value, languages=["fr"]) elif "Date de fin de validité" in label: - e["expiration_date"] = value + e["expiration_date"] = parse_date(value, languages=["fr"]) elif "Catégorie" in label: # TODO: translate? e["category"] = value @@ -386,19 +376,20 @@ def get_germany_certified( # noqa: C901 :param artifacts: Whether to download and compute artifact hashes (way slower, even more data). :return: The entries. """ - base_soup = _get_page(constants.CC_BSI_CERTIFIED_URL) + session = requests.Session() + base_soup = _get_page(constants.CC_BSI_CERTIFIED_URL, session=session) category_nav = base_soup.find("ul", class_="no-bullet row") results = [] for li in tqdm(category_nav.find_all("li"), desc="Get DE scheme certified."): a = li.find("a") url = a["href"] category_name = sns(a.text) - soup = _get_page(urljoin(constants.CC_BSI_BASE_URL, url)) + soup = _get_page(urljoin(constants.CC_BSI_BASE_URL, url), session=session) content = soup.find("div", class_="content").find("div", class_="column") - for table in tqdm(content.find_all("table")): + for table in tqdm(content.find_all("table"), leave=False): tbody = table.find("tbody") header = table.find_parent("div", class_="wrapperTable").find_previous_sibling("h2") - for tr in tqdm(tbody.find_all("tr")): + for tr in tqdm(tbody.find_all("tr"), leave=False): tds = tr.find_all("td") if len(tds) != 4: continue @@ -406,13 +397,13 @@ def get_germany_certified( # noqa: C901 "cert_id": sns(tds[0].text), "product": sns(tds[1].text), "vendor": sns(tds[2].text), - "certification_date": sns(tds[3].text), + "certification_date": parse_date(sns(tds[3].text), "%d.%m.%Y"), "category": category_name, "url": urljoin(constants.CC_BSI_BASE_URL, tds[0].find("a")["href"]), } if enhanced: e: dict[str, Any] = {} - cert_page = _get_page(cert["url"]) + cert_page = _get_page(cert["url"], session=session) content = cert_page.find("div", id="content").find("div", class_="column") head = content.find("h1", class_="c-intro__headline") e["product"] = sns(head.next_sibling.text) @@ -432,9 +423,9 @@ def get_germany_certified( # noqa: C901 elif "Protection Profile" in title: e["protection_profile"] = value elif "Certification Date" in title: - e["certification_date"] = value + e["certification_date"] = parse_date(value, "%d.%m.%Y") elif "valid until" in title: - e["expiration_date"] = value + e["expiration_date"] = parse_date(value, "%d.%m.%Y") links = content.find("ul") if links: # has multiple entries/recertifications @@ -462,15 +453,15 @@ def get_germany_certified( # noqa: C901 if "Certification Report" in title: e["report_link"] = href if artifacts: - e["report_hash"] = _get_hash(href) + e["report_hash"] = _get_hash(href, session=session) elif "Security Target" in title: e["target_link"] = href if artifacts: - e["target_hash"] = _get_hash(href) + e["target_hash"] = _get_hash(href, session=session) elif "Certificate" in title: e["cert_link"] = href if artifacts: - e["cert_hash"] = _get_hash(href) + e["cert_hash"] = _get_hash(href, session=session) description = content.find("div", attrs={"lang": "en"}) if description: e["description"] = sns(description.text) @@ -503,13 +494,14 @@ def get_india_certified() -> list[dict[str, Any]]: # Update pages pager = soup.find("ul", class_="pager__items") - for li in pager.find_all("li"): - try: - new_page = int(li.text) - 1 - except Exception: - continue - if new_page not in seen_pages: - pages.add(new_page) + if pager: + for li in pager.find_all("li"): + try: + new_page = int(li.find("a")["href"].split("=")[1]) + except Exception: + continue + if new_page not in seen_pages: + pages.add(new_page) # Parse table tbody = soup.find("div", class_="view-content").find("table").find("tbody") @@ -526,7 +518,7 @@ def get_india_certified() -> list[dict[str, Any]]: "sponsor": sns(tds[2].text), "developer": sns(tds[3].text), "level": sns(tds[4].text), - "issuance_date": sns(tds[5].text), + "certification_date": parse_date(sns(tds[5].text), "%d-%b-%Y"), "report_link": urljoin(constants.CC_INDIA_BASE_URL, _fix_india_link(report_a["href"])), "report_name": sns(report_a.text), "target_link": urljoin(constants.CC_INDIA_BASE_URL, _fix_india_link(target_a["href"])), @@ -561,7 +553,7 @@ def get_india_archived() -> list[dict[str, Any]]: if pager: for li in pager.find_all("li"): try: - new_page = int(li.text) - 1 + new_page = int(li.find("a")["href"].split("=")[1]) except Exception: continue if new_page not in seen_pages: @@ -586,7 +578,7 @@ def get_india_archived() -> list[dict[str, Any]]: "target_name": sns(target_a.text), "cert_link": urljoin(constants.CC_INDIA_BASE_URL, _fix_india_link(cert_a["href"])), "cert_name": sns(cert_a.text), - "certification_date": sns(tds[8].text), + "certification_date": parse_date(sns(tds[8].text), "%d/%b/%Y"), } if report_a: cert["report_link"] = urljoin(constants.CC_INDIA_BASE_URL, _fix_india_link(report_a["href"])) @@ -615,16 +607,16 @@ def get_italy_certified() -> list[dict[str, Any]]: # noqa: C901 if not p_text or ":" not in p_text: continue p_name, p_data = p_text.split(":") - p_data = p_data + p_data = p_data.strip() p_link = data_p.find("a") if "Fornitore" in p_name: cert["supplier"] = p_data elif "Livello di garanzia" in p_name: cert["level"] = p_data elif "Data emissione certificato" in p_name: - cert["certification_date"] = p_data + cert["certification_date"] = parse_date(p_data, languages=["it"]) elif "Data revisione" in p_name: - cert["revision_date"] = p_data + cert["revision_date"] = parse_date(p_data, languages=["it"]) elif "Rapporto di Certificazione" in p_name and p_link: cert["report_link_it"] = urljoin(constants.CC_ITALY_BASE_URL, p_link["href"]) elif "Certification Report" in p_name and p_link: @@ -669,7 +661,8 @@ def get_italy_in_evaluation() -> list[dict[str, Any]]: def _get_japan(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # noqa: C901 - soup = _get_page(url) + session = requests.Session() + soup = _get_page(url, session=session) table = soup.find("table", class_="cert-table") results = [] trs = list(table.find_all("tr")) @@ -678,16 +671,19 @@ def _get_japan(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # noqa: if not tds: continue if len(tds) in (6, 7): + cert_id = sns(tds[0].text) + if cert_id: + cert_id = "JISEC-CC-CRP-" + cert_id cert: dict[str, Any] = { - "cert_id": sns(tds[0].text), + "cert_id": cert_id, "supplier": sns(tds[1].text), "toe_overseas_name": sns(tds[2].text), } if len(tds) == 6: - cert["expiration_date"] = sns(tds[5].text) + cert["expiration_date"] = parse_date(sns(tds[5].text), "%Y-%m") cert["claim"] = sns(tds[4].text) else: - cert["expiration_date"] = sns(tds[4].text) + cert["expiration_date"] = parse_date(sns(tds[4].text), "%Y-%m") cert["claim"] = sns(tds[5].text) cert_date = sns(tds[3].text) toe_a = tds[2].find("a") @@ -698,7 +694,7 @@ def _get_japan(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # noqa: if cert_date and "Assurance Continuity" in cert_date: cert["revalidations"] = [{"date": cert_date.split("(")[0], "link": toe_link}] else: - cert["certification_date"] = cert_date + cert["certification_date"] = parse_date(cert_date, "%Y-%m") cert["toe_overseas_link"] = toe_link results.append(cert) if len(tds) == 1: @@ -709,7 +705,7 @@ def _get_japan(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # noqa: cert["toe_japan_link"] = urljoin(constants.CC_JAPAN_CERT_BASE_URL, toe_a["href"]) if len(tds) == 2: cert = results[-1] - cert["certification_date"] = sns(tds[1].text) + cert["certification_date"] = parse_date(sns(tds[1].text), "%Y-%m") toe_a = tds[0].find("a") if toe_a and "href" in toe_a.attrs: toe_link = urljoin(constants.CC_JAPAN_CERT_BASE_URL, toe_a["href"]) @@ -717,12 +713,12 @@ def _get_japan(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # noqa: toe_link = None cert["toe_overseas_link"] = toe_link if enhanced: - for cert in results: + for cert in tqdm(results, desc=f"Get JP scheme {name} (enhanced)."): e: dict[str, Any] = {} cert_link = cert.get("toe_overseas_link") or cert.get("toe_japan_link") if not cert_link: continue - cert_page = _get_page(cert_link) + cert_page = _get_page(cert_link, session=session) main = cert_page.find("div", id="main") left = main.find("div", id="left") for dl in left.find_all("dl"): @@ -738,11 +734,18 @@ def _get_japan(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # noqa: elif "Product Type" in title: e["product_type"] = value elif "Certification Identification" in title: - e["cert_id"] = value + if not value or value.startswith("JISEC-CC-CRP-"): + e["cert_id"] = value + elif value.startswith("JISEC-"): + e["cert_id"] = value.replace("JISEC-", "JISEC-CC-CRP-") + else: + e["cert_id"] = value elif "Version of Common Criteria" in title: e["cc_version"] = value - elif "Date" in title: - e["certification_date"] = value + elif "Date of Certification Expiry" in title: + e["expiration_date"] = parse_date(value, "%Y-%m-%d") + elif "Date of Certification" in title: + e["certification_date"] = parse_date(value, "%Y-%m-%d") elif "Conformance Claim" in title: e["assurance_level"] = value elif "PP Identifier" in title and value != "None": @@ -767,15 +770,15 @@ def _get_japan(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # noqa: if "Report" in name: e["report_link"] = urljoin(constants.CC_JAPAN_BASE_URL, li_a["href"]) if artifacts: - e["report_hash"] = _get_hash(e["report_link"]) + e["report_hash"] = _get_hash(e["report_link"], session=session) elif "Certificate" in name: e["cert_link"] = urljoin(constants.CC_JAPAN_BASE_URL, li_a["href"]) if artifacts: - e["cert_hash"] = _get_hash(e["cert_link"]) + e["cert_hash"] = _get_hash(e["cert_link"], session=session) elif "Target" in name: e["target_link"] = urljoin(constants.CC_JAPAN_BASE_URL, li_a["href"]) if artifacts: - e["target_hash"] = _get_hash(e["target_link"]) + e["target_hash"] = _get_hash(e["target_link"], session=session) e["description"] = sns(main.find("div", id="overviewsbox").text) cert["enhanced"] = e return results @@ -830,7 +833,8 @@ def get_japan_in_evaluation() -> list[dict[str, Any]]: def _get_malaysia(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # noqa: C901 - soup = _get_page(url) + session = requests.Session() + soup = _get_page(url, session=session) pages_re = re.search("Page [0-9]+ of ([0-9]+)", soup.find("form").text) if not pages_re: raise ValueError @@ -838,7 +842,7 @@ def _get_malaysia(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # no results = [] pbar = tqdm(desc=f"Get MY scheme {name}.") for i in range(total_pages): - soup = _get_page(url + f"?start={i * 10}") + soup = _get_page(url + f"?start={i * 10}", session=session) table = soup.find("table", class_="directoryTable") for tr in table.find_all("tr", class_="directoryRow"): tds = tr.find_all("td") @@ -847,14 +851,14 @@ def _get_malaysia(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # no "developer": sns(tds[1].text), "level": sns(tds[2].text), "product": sns(tds[3].text), - "certification_date": sns(tds[4].text), - "expiration_date": sns(tds[5].text), + "certification_date": parse_date(sns(tds[4].text), "%d-%m-%Y"), + "expiration_date": parse_date(sns(tds[5].text), "%d-%m-%Y"), "recognition": sns(tds[6].text), "url": urljoin(constants.CC_MALAYSIA_BASE_URL, tds[7].find("a")["href"]), } if enhanced: e: dict[str, Any] = {} - cert_page = _get_page(cert["url"]) + cert_page = _get_page(cert["url"], session=session) for row in cert_page.find_all("div", class_="rsform-table-row"): left = row.find("div", class_="rsform-left-col") right = row.find("div", class_="rsform-right-col") @@ -877,9 +881,9 @@ def _get_malaysia(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # no elif "Assurance Level" in title: e["assurance_level"] = value elif "Certificate Date" in title: - e["certification_date"] = value + e["certification_date"] = parse_date(value, "%d-%m-%Y") elif "Expiry Date" in title: - e["expiration_date"] = value + e["expiration_date"] = parse_date(value, "%d-%m-%Y") elif "Recognized By" in title: e["mutual_recognition"] = value elif "Reports" in title: @@ -887,11 +891,11 @@ def _get_malaysia(url, enhanced, artifacts, name) -> list[dict[str, Any]]: # no if "ST" in a.text: e["target_link"] = urljoin(constants.CC_MALAYSIA_BASE_URL, a["href"]) if artifacts: - e["target_hash"] = _get_hash(e["target_link"]) + e["target_hash"] = _get_hash(e["target_link"], session=session) elif "CR" in a.text: e["report_link"] = urljoin(constants.CC_MALAYSIA_BASE_URL, a["href"]) if artifacts: - e["report_hash"] = _get_hash(e["report_link"]) + e["report_hash"] = _get_hash(e["report_link"], session=session) elif "Maintenance" in title: pass elif "Status" in title: @@ -961,11 +965,15 @@ def _get_netherlands_certified_old( # noqa: C901 for row, modal in tqdm(zip(rows, modals), desc="Get NL scheme certified (old)."): row_entries = row.find_all("a") modal_trs = modal.find_all("tr") + cert_id = sns(row_entries[3].text) + if cert_id: + cert_id = cert_id if cert_id.startswith("NSCIB-") else "NSCIB-" + cert_id + cert_id = cert_id if cert_id.endswith("-CR") else cert_id + "-CR" cert: dict[str, Any] = { "manufacturer": sns(row_entries[0].text), "product": sns(row_entries[1].text), "scheme": sns(row_entries[2].text), - "cert_id": sns(row_entries[3].text), + "cert_id": cert_id, } for tr in modal_trs: th_text = tr.find("th").text @@ -1002,9 +1010,12 @@ def _get_netherlands_certified_new( # noqa: C901 results = [] for tr in tqdm(table.find_all("tr")[1:], desc="Get NL scheme certified (new)."): tds = tr.find_all("td") + cert_id = sns(tds[0].text).replace("\n", "") # type: ignore + cert_id = cert_id if cert_id.startswith("NSCIB-") else "NSCIB-" + cert_id + cert_id = cert_id if cert_id.endswith("-CR") else cert_id + "-CR" cert = { - "cert_id": sns(tds[0].text).replace("\n", ""), # type: ignore - "certification_date": sns(tds[1].text), + "cert_id": cert_id, + "certification_date": parse_date(sns(tds[1].text), "%Y-%m-%d"), "status": sns(tds[2].text), "product": sns(tds[3].text), "developer": sns(tds[4].text), @@ -1082,7 +1093,8 @@ def get_netherlands_in_evaluation() -> list[dict[str, Any]]: def _get_norway( # noqa: C901 url: str, enhanced: bool, artifacts: bool, name ) -> list[dict[str, Any]]: - soup = _get_page(url) + session = requests.Session() + soup = _get_page(url, session=session) results = [] for tr in tqdm(soup.find_all("tr", class_="certified-product"), desc=f"Get NO scheme {name}."): tds = tr.find_all("td") @@ -1091,11 +1103,11 @@ def _get_norway( # noqa: C901 "url": tds[0].find("a")["href"], "category": sns(tds[1].find("p", class_="value").text), "developer": sns(tds[2].find("p", class_="value").text), - "certification_date": sns(tds[3].find("time").text), + "certification_date": parse_date(sns(tds[3].find("time").text), "%d.%m.%Y"), } if enhanced: e: dict[str, Any] = {} - cert_page = _get_page(cert["url"]) + cert_page = _get_page(cert["url"], session=session) content = cert_page.find("div", class_="main-content") body = content.find("div", class_="articleelement") if body: @@ -1107,7 +1119,8 @@ def _get_norway( # noqa: C901 if not title: continue if "Certificate No." in title and value is not None: - e["id"] = value.split(" ")[0] + value = value.split(" ")[0] + e["cert_id"] = value if value.startswith("SERTIT-") else "SERTIT-" + value elif "Mutual Recognition" in title: e["mutual_recognition"] = value elif "Product" in title: @@ -1121,7 +1134,9 @@ def _get_norway( # noqa: C901 elif "Evaluation Facility" in title: e["evaluation_facility"] = value elif "Certification Date" in title: - e["certification_date"] = value + e["certification_date"] = parse_date(value, "%d.%m.%Y") + elif "Certificate Expiration Date" in title: + e["expiration_date"] = parse_date(value, "%d.%m.%Y") elif "Evaluation Level" in title: e["level"] = value elif "Protection Profile" in title: @@ -1148,7 +1163,7 @@ def _get_norway( # noqa: C901 a = link.find("a") entry = {"href": urljoin(constants.CC_NORWAY_BASE_URL, a["href"])} if artifacts: - entry["hash"] = _get_hash(entry["href"]) # type: ignore + entry["hash"] = _get_hash(entry["href"], session=session) # type: ignore entries.append(entry) e["documents"][doc_type] = entries cert["enhanced"] = e @@ -1179,12 +1194,12 @@ def get_norway_archived(enhanced: bool = True, artifacts: bool = False) -> list[ def _get_korea( # noqa: C901 - product_class: int, enhanced: bool, artifacts: bool, name + url: str, product_class: int, enhanced: bool, artifacts: bool, name ) -> list[dict[str, Any]]: - session = requests.session() + session = requests.Session() _get_page(constants.CC_KOREA_EN_URL, session=session) # Get base page - url = constants.CC_KOREA_CERTIFIED_URL + f"?product_class={product_class}" + url = url + f"?product_class={product_class}" soup = _get_page(url, session=session) seen_pages = set() pages = {1} @@ -1206,14 +1221,17 @@ def _get_korea( # noqa: C901 continue link = tds[0].find("a") id = link["id"].split("-")[1] + cert_id = sns(tds[1].text) + if cert_id: + cert_id = cert_id.replace(" ", "-") cert: dict[str, Any] = { "product": sns(tds[0].text), - "cert_id": sns(tds[1].text), - "product_link": constants.CC_KOREA_PRODUCT_URL.format(id), + "cert_id": cert_id, + "product_link": constants.CC_KOREA_PRODUCT_URL.format(id, product_class), "vendor": sns(tds[2].text), "level": sns(tds[3].text), "category": sns(tds[4].text), - "certification_date": sns(tds[5].text), + "certification_date": parse_date(sns(tds[5].text), "%Y-%m-%d"), } if enhanced: e: dict[str, Any] = {} @@ -1241,11 +1259,11 @@ def _get_korea( # noqa: C901 elif "Common Criteria" in title: v["cc_version"] = value elif "Date of Certification" in title or "Date issued" in title: - v["certification_date"] = value + v["certification_date"] = parse_date(value, "%Y-%m-%d") elif "EvaluationAssurance Level" in title: v["assurance_level"] = value elif "Expiry Date" in title: - v["expiration_date"] = value + v["expiration_date"] = parse_date(value, "%Y-%m-%d") elif "Type of Product" in title: v["product_type"] = value elif "Certification No." in title: @@ -1296,7 +1314,9 @@ def get_korea_certified(enhanced: bool = True, artifacts: bool = False) -> list[ :param artifacts: Whether to download and compute artifact hashes (way slower, even more data). :return: The entries. """ - return _get_korea(product_class=1, enhanced=enhanced, artifacts=artifacts, name="certified") + return _get_korea( + constants.CC_KOREA_CERTIFIED_URL, product_class=1, enhanced=enhanced, artifacts=artifacts, name="certified" + ) def get_korea_suspended(enhanced: bool = True, artifacts: bool = False) -> list[dict[str, Any]]: @@ -1307,7 +1327,9 @@ def get_korea_suspended(enhanced: bool = True, artifacts: bool = False) -> list[ :param artifacts: Whether to download and compute artifact hashes (way slower, even more data). :return: The entries. """ - return _get_korea(product_class=2, enhanced=enhanced, artifacts=artifacts, name="suspended") + return _get_korea( + constants.CC_KOREA_SUSPENDED_URL, product_class=2, enhanced=enhanced, artifacts=artifacts, name="suspended" + ) def get_korea_archived(enhanced: bool = True, artifacts: bool = False) -> list[dict[str, Any]]: @@ -1318,7 +1340,9 @@ def get_korea_archived(enhanced: bool = True, artifacts: bool = False) -> list[d :param artifacts: Whether to download and compute artifact hashes (way slower, even more data). :return: The entries. """ - return _get_korea(product_class=4, enhanced=enhanced, artifacts=artifacts, name="archived") + return _get_korea( + constants.CC_KOREA_ARCHIVED_URL, product_class=4, enhanced=enhanced, artifacts=artifacts, name="archived" + ) def get_poland_certified(artifacts: bool = False) -> list[dict[str, Any]]: @@ -1336,8 +1360,8 @@ def get_poland_certified(artifacts: bool = False) -> list[dict[str, Any]]: cert = { "client": sns(tds[0].text), "product": sns(tds[1].text), - "certification_date": sns(tds[4].text), - "expiration_date": sns(tds[5].text), + "certification_date": parse_date(sns(tds[4].text), "%d.%m.%Y"), + "expiration_date": parse_date(sns(tds[5].text), "%d.%m.%Y"), } cc_entry = sns(tds[2].text) cc_split = None @@ -1364,8 +1388,6 @@ def get_poland_ineval() -> list[dict[str, Any]]: """ Get Polish "product in evaluation" entries. - :param enhanced: Whether to enhance the results by following links (slower, more data). - :param artifacts: Whether to download and compute artifact hashes (way slower, even more data). :return: The entries. """ soup = _get_page(constants.CC_POLAND_INEVAL_URL) @@ -1391,13 +1413,14 @@ def get_poland_ineval() -> list[dict[str, Any]]: def _get_singapore(url: str, artifacts: bool, name) -> list[dict[str, Any]]: - soup = _get_page(url) + session = requests.Session() + soup = _get_page(url, session=session) page_id = str(soup.find("input", id="CurrentPageId").value) page = 1 - api_call = requests.post( + api_call = session.post( constants.CC_SINGAPORE_API_URL, data={ - "PassSortFilter": False, + "PassSortFilter": name == "archived", "currentPageId": page_id, "page": page, "limit": 15, @@ -1415,8 +1438,8 @@ def _get_singapore(url: str, artifacts: bool, name) -> list[dict[str, Any]]: "product": obj["productName"], "vendor": obj["productDeveloper"], "url": urljoin(constants.CC_SINGAPORE_BASE_URL, obj["productUrl"]), - "certification_date": obj["dateOfIssuance"], - "expiration_date": obj["dateOfExpiry"], + "certification_date": parse_date(obj["dateOfIssuance"], "%d %B %Y"), + "expiration_date": parse_date(obj["dateOfExpiry"], "%d %B %Y"), "category": obj["productCategory"]["title"], "cert_title": obj["certificate"]["title"], "cert_link": urljoin(constants.CC_SINGAPORE_BASE_URL, obj["certificate"]["mediaUrl"]), @@ -1435,10 +1458,10 @@ def _get_singapore(url: str, artifacts: bool, name) -> list[dict[str, Any]]: pbar.update() results.append(cert) page += 1 - api_call = requests.post( + api_call = session.post( constants.CC_SINGAPORE_API_URL, data={ - "PassSortFilter": False, + "PassSortFilter": name == "archived", "currentPageId": page_id, "page": page, "limit": 15, @@ -1496,13 +1519,15 @@ def get_singapore_archived(artifacts: bool = False) -> list[dict[str, Any]]: return _get_singapore(constants.CC_SINGAPORE_ARCHIVED_URL, artifacts, "archived") -def get_spain_certified() -> list[dict[str, Any]]: +def get_spain_certified(enhanced: bool = True) -> list[dict[str, Any]]: # noqa: C901 """ Get Spanish "certified product" entries. + :param enhanced: Whether to enhance the results by following links (slower, more data). :return: The entries. """ - soup = _get_page(constants.CC_SPAIN_CERTIFIED_URL) + session = requests.Session() + soup = _get_page(constants.CC_SPAIN_CERTIFIED_URL, session=session) tbody = soup.find("table", class_="djc_items_table").find("tbody") results = [] for tr in tqdm(tbody.find_all("tr", recursive=False), desc="Get ES scheme certified."): @@ -1512,8 +1537,50 @@ 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].text), + "certification_date": parse_date(sns(tds[3].text), "%d/%m/%Y"), } + if enhanced: + e: dict[str, Any] = {} + if not cert["product_link"]: + continue + cert_page = _get_page(cert["product_link"], session=session) + description_div = cert_page.find("div", class_="djc_description") + e["description"] = sns(description_div.find("div", class_="djc_desc_wrap").text) + category_a = description_div.find("div", class_="djc_category_info").find("a") + if category_a: + e["category"] = sns(category_a.text) + e["manufacturer"] = sns(description_div.find("div", class_="djc_producer_info").find("span").text) + for attr in description_div.find_all("p", class_="djc_attribute"): + label_text = sns(attr.find("span", class_="djc_attribute-label").text) + value = sns(attr.find("span", class_="djc_value").text) + if not label_text: + continue + if "Type" in label_text: + e["type"] = value + elif "Testing laboratory" in label_text: + e["evaluation_facility"] = value + elif "Certification Status" in label_text: + e["status"] = value + elif "Certification Date" in label_text: + e["certification_date"] = parse_date(value, "%d-%m-%Y") + elif "Standard Version" in label_text: + e["cc_version"] = value + elif "Evaluation Level" in label_text: + e["level"] = value + for file in description_div.find_all("p", class_="djc_file"): + label_text = sns(file.find("span", class_="djc_att_group_label").text) + if not label_text: + continue + if "CCRA Certificate" in label_text: + file_type = "cert" + elif "Security Target" in label_text: + file_type = "target" + elif "Certification Report" in label_text: + file_type = "report" + else: + continue + e[f"{file_type}_link"] = urljoin(constants.CC_SPAIN_BASE_URL, file.find("a")["href"]) + cert["enhanced"] = e results.append(cert) return results @@ -1521,7 +1588,8 @@ def get_spain_certified() -> list[dict[str, Any]]: def _get_sweden( # noqa: C901 url: str, enhanced: bool, artifacts: bool, name ) -> list[dict[str, Any]]: - soup = _get_page(url) + session = requests.Session() + soup = _get_page(url, session=session) nav = soup.find("main").find("nav", class_="component-nav-box__list") results = [] for link in tqdm(nav.find_all("a"), desc=f"Get SE scheme {name}."): @@ -1533,7 +1601,7 @@ def _get_sweden( # noqa: C901 e: dict[str, Any] = {} if not cert["url"]: continue - cert_page = _get_page(cert["url"]) + cert_page = _get_page(cert["url"], session=session) content = cert_page.find("section", class_="container-article") head = content.find("h1") e["title"] = sns(head.text) @@ -1559,7 +1627,7 @@ def _get_sweden( # noqa: C901 elif "Assuranspaket" in title: e["assurance_level"] = value elif "Certifieringsdatum" in title: - e["certification_date"] = value + e["certification_date"] = parse_date(value, "%Y-%m-%d") elif "Sponsor" in title: e["sponsor"] = value elif "Utvecklare" in title: @@ -1569,15 +1637,15 @@ def _get_sweden( # noqa: C901 elif "Security Target" in title and a: e["target_link"] = urljoin(constants.CC_SWEDEN_BASE_URL, a["href"]) if artifacts: - e["target_hash"] = _get_hash(e["target_link"]) + e["target_hash"] = _get_hash(e["target_link"], session=session) elif "Certifieringsrapport" in title and a: e["report_link"] = urljoin(constants.CC_SWEDEN_BASE_URL, a["href"]) if artifacts: - e["report_hash"] = _get_hash(e["report_hash"]) + e["report_hash"] = _get_hash(e["report_hash"], session=session) elif "Certifikat" in title and a: e["cert_link"] = urljoin(constants.CC_SWEDEN_BASE_URL, a["href"]) if artifacts: - e["cert_hash"] = _get_hash(e["cert_link"]) + e["cert_hash"] = _get_hash(e["cert_link"], session=session) cert["enhanced"] = e results.append(cert) return results @@ -1640,7 +1708,7 @@ def get_turkey_certified() -> list[dict[str, Any]]: "product": values[2], "cc_version": values[3], "level": values[4], - "cert_lab": values[5], + "evaluation_facility": values[5], "certification_date": values[6], "expiration_date": values[7], # TODO: Parse "Ongoing Evaluation" out of this field as well. @@ -1656,9 +1724,9 @@ def _get_usa(args, enhanced: bool, artifacts: bool, name): # 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"], + "url": urljoin(constants.CC_USA_BASE_URL, f"/product/{cert['product_id']}"), + "certification_date": parse_date(cert["certification_date"], "%m/%d/%Y"), + "expiration_date": parse_date(cert["sunset_date"], "%m/%d/%Y"), "category": cert["tech_type"], "vendor": cert["vendor_id_name"], "evaluation_facility": cert["assigned_lab_name"], @@ -1691,7 +1759,7 @@ def _get_usa(args, enhanced: bool, artifacts: bool, name): # noqa: C901 return result - session = requests.session() + session = requests.Session() results = [] offset = 0 got = 0 @@ -1817,7 +1885,7 @@ class CCScheme(ComplexSerializableType): EntryType.Certified: get_norway_certified, EntryType.Archived: get_norway_archived, }, - "KO": { + "KR": { EntryType.Certified: get_korea_certified, EntryType.Archived: get_korea_archived, }, @@ -1846,10 +1914,24 @@ class CCScheme(ComplexSerializableType): @classmethod def from_dict(cls, dct): + def _deserialize_entry(entry): + if isinstance(entry, dict): + res = {} + for key, value in entry.items(): + if key.endswith("_date"): + res[key] = date.fromisoformat(value) if value is not None else None + else: + res[key] = _deserialize_entry(value) + return res + elif isinstance(entry, list): + return list(map(_deserialize_entry, entry)) + else: + return entry + return cls( dct["country"], datetime.fromisoformat(dct["timestamp"]), - {EntryType(entry_type): entries for entry_type, entries in dct["lists"].items()}, + {EntryType(entry_type): _deserialize_entry(entries) for entry_type, entries in dct["lists"].items()}, ) def to_dict(self): diff --git a/src/sec_certs/sample/certificate.py b/src/sec_certs/sample/certificate.py index 5dab9cef..74b1af96 100644 --- a/src/sec_certs/sample/certificate.py +++ b/src/sec_certs/sample/certificate.py @@ -64,7 +64,7 @@ class Certificate(Generic[T, H, P], ABC, ComplexSerializableType): @property @abstractmethod - def dgst(self): + def dgst(self) -> str: raise NotImplementedError("Not meant to be implemented") @property diff --git a/src/sec_certs/utils/extract.py b/src/sec_certs/utils/extract.py index 340ad568..6af7f4cc 100644 --- a/src/sec_certs/utils/extract.py +++ b/src/sec_certs/utils/extract.py @@ -741,29 +741,6 @@ def load_text_file( def rules_get_subset(desired_path: str) -> dict: """ - - - - - - - - - - - - - - - - - - - - - - - Recursively applies cc_certs.get(key) on tokens from desired_path, returns the keys of the inner-most layer. """ diff --git a/src/sec_certs/utils/helpers.py b/src/sec_certs/utils/helpers.py index dedee959..6a6a9954 100644 --- a/src/sec_certs/utils/helpers.py +++ b/src/sec_certs/utils/helpers.py @@ -11,6 +11,7 @@ from functools import partial from pathlib import Path from typing import Any +import dateparser import numpy as np import pkgconfig import requests @@ -21,7 +22,6 @@ from sec_certs.utils.tqdm import tqdm logger = logging.getLogger(__name__) - _PROXIES = { "https://www.commoncriteriaportal.org/": "https://sec-certs.org/proxy/cc/", "https://csrc.nist.gov/": "https://sec-certs.org/proxy/fips/", @@ -120,6 +120,20 @@ def to_utc(timestamp: datetime) -> datetime: return timestamp.replace(tzinfo=None) +def parse_date(date_string: str | None, format: str | None = None, languages: list[str] | None = None): + if not date_string: + return None + if format: + try: + return datetime.strptime(date_string, format).date() + except ValueError: + pass + parsed = dateparser.parse(date_string, languages=languages) + if parsed is not None: + return parsed.date() + return None + + def is_in_dict(target_dict: dict, path: str) -> bool: current_level = target_dict for item in path: diff --git a/src/sec_certs/utils/sanitization.py b/src/sec_certs/utils/sanitization.py index 4a7326eb..9bf4ad95 100644 --- a/src/sec_certs/utils/sanitization.py +++ b/src/sec_certs/utils/sanitization.py @@ -2,6 +2,7 @@ from __future__ import annotations import html import logging +import re from datetime import date from pathlib import Path from urllib.parse import urlparse @@ -16,7 +17,9 @@ logger = logging.getLogger(__name__) def sanitize_navigable_string(string: NavigableString | str | None) -> str | None: if not string: return None - return str(string).strip().replace("\xad", "").replace("\xa0", "") + rex = re.compile(r"\s+") + string = str(string).strip().replace("\xad", "").replace("\xa0", "") + return rex.sub(" ", string) def sanitize_link(record: str | None) -> str | None: diff --git a/src/sec_certs/utils/tqdm.py b/src/sec_certs/utils/tqdm.py index 581295ad..0a23903f 100644 --- a/src/sec_certs/utils/tqdm.py +++ b/src/sec_certs/utils/tqdm.py @@ -1,4 +1,4 @@ -from tqdm import tqdm as tqdm_original +from tqdm.auto import tqdm as tqdm_original from sec_certs.configuration import config |
