aboutsummaryrefslogtreecommitdiffhomepage
path: root/src
diff options
context:
space:
mode:
authoradamjanovsky2024-02-20 09:55:01 +0100
committeradamjanovsky2024-02-20 09:55:01 +0100
commit6f8a09d7696f3309455246f6e2b9623ecdd34d27 (patch)
treed041d31b72155972bab72e688014c797d19207db /src
parent89da697f14f3eab13f4836722281d9a723ec0b6a (diff)
parent609ee028fad536c7a7e3d83b004b0617ebaaf460 (diff)
downloadsec-certs-6f8a09d7696f3309455246f6e2b9623ecdd34d27.tar.gz
sec-certs-6f8a09d7696f3309455246f6e2b9623ecdd34d27.tar.zst
sec-certs-6f8a09d7696f3309455246f6e2b9623ecdd34d27.zip
Merge branch 'main' into reference-notebook
Diffstat (limited to 'src')
-rw-r--r--src/sec_certs/configuration.py13
-rw-r--r--src/sec_certs/constants.py10
-rw-r--r--src/sec_certs/dataset/cc.py136
-rw-r--r--src/sec_certs/dataset/fips.py4
-rw-r--r--src/sec_certs/model/cc_matching.py4
-rw-r--r--src/sec_certs/model/references_nlp/segment_extractor.py8
-rw-r--r--src/sec_certs/rules.yaml220
-rw-r--r--src/sec_certs/sample/cc.py600
-rw-r--r--src/sec_certs/sample/cc_certificate_id.py325
-rw-r--r--src/sec_certs/sample/cc_scheme.py369
-rw-r--r--src/sec_certs/sample/cve.py2
-rw-r--r--src/sec_certs/sample/fips_mip.py1
-rw-r--r--src/sec_certs/sample/sar.py2
-rw-r--r--src/sec_certs/utils/extract.py9
14 files changed, 1044 insertions, 659 deletions
diff --git a/src/sec_certs/configuration.py b/src/sec_certs/configuration.py
index 3ebb22bd..81fac197 100644
--- a/src/sec_certs/configuration.py
+++ b/src/sec_certs/configuration.py
@@ -6,7 +6,7 @@ from typing import Literal, Optional
import yaml
from pydantic import AnyHttpUrl, Field
-from pydantic_settings import BaseSettings
+from pydantic_settings import BaseSettings, SettingsConfigDict
class Configuration(BaseSettings):
@@ -15,8 +15,7 @@ class Configuration(BaseSettings):
While not a singleton, the `config` instance from this module is meant to be primarily used.
"""
- class Config:
- env_prefix = "seccerts_"
+ model_config = SettingsConfigDict(env_prefix="seccerts_")
log_filepath: Path = Field(
"./cert_processing_log.log",
@@ -122,12 +121,12 @@ class Configuration(BaseSettings):
"""
Returns keys of the config that have non-default value, i.e. were provided as kwargs, env. vars. or additionaly set.
"""
- return {key for key, value in Configuration.__fields__.items() if getattr(self, key) != value.default}
+ return {key for key, value in Configuration.model_fields.items() if getattr(self, key) != value.default}
def _set_attrs_from_cfg(self, other_cfg: Configuration, fields_to_set: set[str] | None) -> None:
if not fields_to_set:
- fields_to_set = set(Configuration.__fields__.keys())
- for field in [x for x in other_cfg.__fields__ if x in fields_to_set]:
+ fields_to_set = set(Configuration.model_fields.keys())
+ for field in [x for x in other_cfg.model_fields if x in fields_to_set]:
setattr(self, field, getattr(other_cfg, field))
def load_from_yaml(self, yaml_path: str | Path) -> None:
@@ -139,7 +138,7 @@ class Configuration(BaseSettings):
"""
with Path(yaml_path).open("r") as handle:
data = yaml.safe_load(handle)
- other_cfg = Configuration.parse_obj(data)
+ other_cfg = Configuration.model_validate(data)
keys_to_rewrite = set(data.keys()).union(other_cfg._get_nondefault_keys())
self._set_attrs_from_cfg(other_cfg, keys_to_rewrite)
diff --git a/src/sec_certs/constants.py b/src/sec_certs/constants.py
index 01649bb7..620f984f 100644
--- a/src/sec_certs/constants.py
+++ b/src/sec_certs/constants.py
@@ -82,8 +82,9 @@ CC_CANADA_BASE_URL = "https://www.cyber.gc.ca"
CC_CANADA_API_URL = CC_CANADA_BASE_URL + "/api/cccs/page/v1/get"
CC_CANADA_CERTIFIED_URL = "/en/tools-services/common-criteria/certified-products"
CC_CANADA_INEVAL_URL = "/en/tools-services/common-criteria/products-evaluation"
-CC_ANSSI_BASE_URL = "https://www.ssi.gouv.fr"
-CC_ANSSI_CERTIFIED_URL = CC_ANSSI_BASE_URL + "/en/products/certified-products/"
+CC_ANSSI_BASE_URL = "https://cyber.gouv.fr"
+CC_ANSSI_CERTIFIED_URL = CC_ANSSI_BASE_URL + "/produits-certifies"
+CC_ANSSI_ARCHIVED_URL = CC_ANSSI_BASE_URL + "/produits-certifies-archives"
CC_BSI_BASE_URL = "https://www.bsi.bund.de/"
CC_BSI_CERTIFIED_URL = (
CC_BSI_BASE_URL
@@ -103,7 +104,10 @@ CC_JAPAN_ARCHIVED_SW_URL = CC_JAPAN_BASE_URL + "/software/certified-cert/archive
CC_JAPAN_INEVAL_URL = CC_JAPAN_BASE_URL + "/prdct-in-eval/in_eval_list.html"
CC_MALAYSIA_BASE_URL = "https://iscb.cybersecurity.my"
CC_MALAYSIA_CERTIFIED_URL = (
- CC_MALAYSIA_BASE_URL + "/index.php/certification/product-certification/mycc/certified-products-and-systems"
+ CC_MALAYSIA_BASE_URL + "/index.php/certification/product-certification/mycc/certified-products-and-systems-5"
+)
+CC_MALAYSIA_ARCHIVED_URL = (
+ CC_MALAYSIA_BASE_URL + "/index.php/certification/product-certification/mycc/archived-certified-products-and-systems"
)
CC_MALAYSIA_INEVAL_URL = (
CC_MALAYSIA_BASE_URL
diff --git a/src/sec_certs/dataset/cc.py b/src/sec_certs/dataset/cc.py
index 04db4d7c..44c56afa 100644
--- a/src/sec_certs/dataset/cc.py
+++ b/src/sec_certs/dataset/cc.py
@@ -83,8 +83,8 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
df = pd.DataFrame([x.pandas_tuple for x in self.certs.values()], columns=CCCertificate.pandas_columns)
df = df.set_index("dgst")
- df.not_valid_before = pd.to_datetime(df.not_valid_before, infer_datetime_format=True, errors="coerce")
- df.not_valid_after = pd.to_datetime(df.not_valid_after, infer_datetime_format=True, errors="coerce")
+ df.not_valid_before = pd.to_datetime(df.not_valid_before, errors="coerce")
+ df.not_valid_after = pd.to_datetime(df.not_valid_after, errors="coerce")
df = df.astype(
{"category": "category", "status": "category", "scheme": "category", "cert_lab": "category"}
).fillna(value=np.nan)
@@ -144,6 +144,27 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
return self.targets_dir / "txt"
@property
+ def certificates_dir(self) -> Path:
+ """
+ Returns directory that holds files associated with the certificates
+ """
+ return self.certs_dir / "certificates"
+
+ @property
+ def certificates_pdf_dir(self) -> Path:
+ """
+ Returns directory that holds PDFs associated with certificates
+ """
+ return self.certificates_dir / "pdf"
+
+ @property
+ def certificates_txt_dir(self) -> Path:
+ """
+ Returns directory that holds TXTs associated with certificates
+ """
+ return self.certificates_dir / "txt"
+
+ @property
def pp_dataset_path(self) -> Path:
"""
Returns a path to the dataset of Protection Profiles
@@ -178,7 +199,7 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
BASE_URL: ClassVar[str] = "https://www.commoncriteriaportal.org"
HTML_PRODUCTS_URL = {
- "cc_products_active.html": BASE_URL + "/products/",
+ "cc_products_active.html": BASE_URL + "/products/index.cfm",
"cc_products_archived.html": BASE_URL + "/products/index.cfm?archived=1",
}
HTML_LABS_URL = {"cc_labs.html": BASE_URL + "/labs"}
@@ -242,7 +263,14 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
self.auxiliary_datasets.mu_dset.root_dir = self.mu_dataset_dir
for cert in self:
- cert.set_local_paths(self.reports_pdf_dir, self.targets_pdf_dir, self.reports_txt_dir, self.targets_txt_dir)
+ cert.set_local_paths(
+ self.reports_pdf_dir,
+ self.targets_pdf_dir,
+ self.certificates_pdf_dir,
+ self.reports_txt_dir,
+ self.targets_txt_dir,
+ self.certificates_txt_dir,
+ )
# TODO: This forgets to set local paths for other auxiliary datasets
def _merge_certs(self, certs: dict[str, CCCertificate], cert_source: str | None = None) -> None:
@@ -362,7 +390,10 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
]
# TODO: Now skipping bad lines, smarter heuristics to be built for dumb files
- df = pd.read_csv(file, engine="python", encoding="windows-1252", on_bad_lines="skip")
+ try:
+ df = pd.read_csv(file, engine="python", encoding="utf-8", on_bad_lines="skip")
+ except UnicodeDecodeError:
+ df = pd.read_csv(file, engine="python", encoding="windows-1252", on_bad_lines="skip")
df = df.rename(columns=dict(zip(list(df.columns), csv_header)))
df["is_maintenance"] = ~df.maintenance_title.isnull()
@@ -528,11 +559,12 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
def _download_all_artifacts_body(self, fresh: bool = True) -> None:
self._download_reports(fresh)
self._download_targets(fresh)
+ self._download_certs(fresh)
@staged(logger, "Downloading PDFs of CC certification reports.")
def _download_reports(self, fresh: bool = True) -> None:
self.reports_pdf_dir.mkdir(parents=True, exist_ok=True)
- certs_to_process = [x for x in self if x.state.report_is_ok_to_download(fresh) and x.report_link]
+ certs_to_process = [x for x in self if x.state.report.is_ok_to_download(fresh) and x.report_link]
if not fresh and certs_to_process:
logger.info(
@@ -548,7 +580,7 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
@staged(logger, "Downloading PDFs of CC security targets.")
def _download_targets(self, fresh: bool = True) -> None:
self.targets_pdf_dir.mkdir(parents=True, exist_ok=True)
- certs_to_process = [x for x in self if x.state.report_is_ok_to_download(fresh)]
+ certs_to_process = [x for x in self if x.state.st.is_ok_to_download(fresh)]
if not fresh and certs_to_process:
logger.info(
@@ -561,10 +593,26 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
progress_bar_desc="Downloading PDFs of CC security targets",
)
+ @staged(logger, "Downloading PDFs of CC certificates.")
+ def _download_certs(self, fresh: bool = True) -> None:
+ self.certificates_pdf_dir.mkdir(parents=True, exist_ok=True)
+ certs_to_process = [x for x in self if x.state.cert.is_ok_to_download(fresh)]
+
+ if not fresh and certs_to_process:
+ logger.info(
+ f"Downloading {len(certs_to_process)} PDFs of CC certificates for which previous download failed.."
+ )
+
+ cert_processing.process_parallel(
+ CCCertificate.download_pdf_cert,
+ certs_to_process,
+ progress_bar_desc="Downloading PDFs of CC certificates",
+ )
+
@staged(logger, "Converting PDFs of certification reports to txt.")
def _convert_reports_to_txt(self, fresh: bool = True) -> None:
self.reports_txt_dir.mkdir(parents=True, exist_ok=True)
- certs_to_process = [x for x in self if x.state.report_is_ok_to_convert(fresh)]
+ certs_to_process = [x for x in self if x.state.report.is_ok_to_convert(fresh)]
if not fresh and certs_to_process:
logger.info(
@@ -580,7 +628,7 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
@staged(logger, "Converting PDFs of security targets to txt.")
def _convert_targets_to_txt(self, fresh: bool = True) -> None:
self.targets_txt_dir.mkdir(parents=True, exist_ok=True)
- certs_to_process = [x for x in self if x.state.st_is_ok_to_convert(fresh)]
+ certs_to_process = [x for x in self if x.state.st.is_ok_to_convert(fresh)]
if fresh:
logger.info("Converting PDFs of security targets to txt.")
@@ -595,13 +643,32 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
progress_bar_desc="Converting PDFs of security targets to txt",
)
+ @staged(logger, "Converting PDFs of certificates to txt.")
+ def _convert_certs_to_txt(self, fresh: bool = True) -> None:
+ self.certificates_txt_dir.mkdir(parents=True, exist_ok=True)
+ certs_to_process = [x for x in self if x.state.cert.is_ok_to_convert(fresh)]
+
+ if fresh:
+ logger.info("Converting PDFs of certificates to txt.")
+ if not fresh and certs_to_process:
+ logger.info(
+ f"Converting {len(certs_to_process)} PDFs of certificates to txt for which previous conversion failed."
+ )
+
+ cert_processing.process_parallel(
+ CCCertificate.convert_cert_pdf,
+ certs_to_process,
+ progress_bar_desc="Converting PDFs of certificates to txt",
+ )
+
def _convert_all_pdfs_body(self, fresh: bool = True) -> None:
self._convert_reports_to_txt(fresh)
self._convert_targets_to_txt(fresh)
+ self._convert_certs_to_txt(fresh)
@staged(logger, "Extracting report metadata")
def _extract_report_metadata(self) -> None:
- certs_to_process = [x for x in self if x.state.report_is_ok_to_analyze()]
+ certs_to_process = [x for x in self if x.state.report.is_ok_to_analyze()]
processed_certs = cert_processing.process_parallel(
CCCertificate.extract_report_pdf_metadata,
certs_to_process,
@@ -612,7 +679,7 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
@staged(logger, "Extracting target metadata")
def _extract_target_metadata(self) -> None:
- certs_to_process = [x for x in self if x.state.st_is_ok_to_analyze()]
+ certs_to_process = [x for x in self if x.state.st.is_ok_to_analyze()]
processed_certs = cert_processing.process_parallel(
CCCertificate.extract_st_pdf_metadata,
certs_to_process,
@@ -621,13 +688,25 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
)
self.update_with_certs(processed_certs)
+ @staged(logger, "Extracting cert metadata")
+ def _extract_cert_metadata(self) -> None:
+ certs_to_process = [x for x in self if x.state.cert.is_ok_to_analyze()]
+ processed_certs = cert_processing.process_parallel(
+ CCCertificate.extract_cert_pdf_metadata,
+ certs_to_process,
+ use_threading=False,
+ progress_bar_desc="Extracting cert metadata",
+ )
+ self.update_with_certs(processed_certs)
+
def _extract_pdf_metadata(self) -> None:
self._extract_report_metadata()
self._extract_target_metadata()
+ self._extract_cert_metadata()
@staged(logger, "Extracting report frontpages")
def _extract_report_frontpage(self) -> None:
- certs_to_process = [x for x in self if x.state.report_is_ok_to_analyze()]
+ certs_to_process = [x for x in self if x.state.report.is_ok_to_analyze()]
processed_certs = cert_processing.process_parallel(
CCCertificate.extract_report_pdf_frontpage,
certs_to_process,
@@ -636,24 +715,13 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
)
self.update_with_certs(processed_certs)
- @staged(logger, "Extracting target frontpages")
- def _extract_target_frontpage(self) -> None:
- certs_to_process = [x for x in self if x.state.st_is_ok_to_analyze()]
- processed_certs = cert_processing.process_parallel(
- CCCertificate.extract_st_pdf_frontpage,
- certs_to_process,
- use_threading=False,
- progress_bar_desc="Extracting target frontpages",
- )
- self.update_with_certs(processed_certs)
-
def _extract_pdf_frontpage(self) -> None:
self._extract_report_frontpage()
- self._extract_target_frontpage()
+ # We have no frontpage extraction for targets or certificates themselves, only for the reports.
@staged(logger, "Extracting report keywords")
def _extract_report_keywords(self) -> None:
- certs_to_process = [x for x in self if x.state.report_is_ok_to_analyze()]
+ certs_to_process = [x for x in self if x.state.report.is_ok_to_analyze()]
processed_certs = cert_processing.process_parallel(
CCCertificate.extract_report_pdf_keywords,
certs_to_process,
@@ -664,7 +732,7 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
@staged(logger, "Extracting target keywords")
def _extract_target_keywords(self) -> None:
- certs_to_process = [x for x in self if x.state.st_is_ok_to_analyze()]
+ certs_to_process = [x for x in self if x.state.st.is_ok_to_analyze()]
processed_certs = cert_processing.process_parallel(
CCCertificate.extract_st_pdf_keywords,
certs_to_process,
@@ -673,9 +741,21 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
)
self.update_with_certs(processed_certs)
+ @staged(logger, "Extracting cert keywords")
+ def _extract_cert_keywords(self) -> None:
+ certs_to_process = [x for x in self if x.state.cert.is_ok_to_analyze()]
+ processed_certs = cert_processing.process_parallel(
+ CCCertificate.extract_cert_pdf_keywords,
+ certs_to_process,
+ use_threading=False,
+ progress_bar_desc="Extracting cert keywords",
+ )
+ self.update_with_certs(processed_certs)
+
def _extract_pdf_keywords(self) -> None:
self._extract_report_keywords()
self._extract_target_keywords()
+ self._extract_cert_keywords()
def extract_data(self) -> None:
logger.info("Extracting various data from certification artifacts")
@@ -685,7 +765,7 @@ class CCDataset(Dataset[CCCertificate, CCAuxiliaryDatasets], ComplexSerializable
@staged(logger, "Computing heuristics: Deriving information about laboratories involved in certification.")
def _compute_cert_labs(self) -> None:
- certs_to_process = [x for x in self if x.state.report_is_ok_to_analyze()]
+ certs_to_process = [x for x in self if x.state.report.is_ok_to_analyze()]
for cert in certs_to_process:
cert.compute_heuristics_cert_lab()
@@ -902,7 +982,7 @@ class CCDatasetMaintenanceUpdates(CCDataset, ComplexSerializableType):
df = df.set_index("dgst")
df.index.name = "dgst"
- df.maintenance_date = pd.to_datetime(df.maintenance_date, infer_datetime_format=True, errors="coerce")
+ df.maintenance_date = pd.to_datetime(df.maintenance_date, errors="coerce")
return df.fillna(value=np.nan)
@classmethod
diff --git a/src/sec_certs/dataset/fips.py b/src/sec_certs/dataset/fips.py
index a3d24b15..77f38754 100644
--- a/src/sec_certs/dataset/fips.py
+++ b/src/sec_certs/dataset/fips.py
@@ -326,8 +326,8 @@ class FIPSDataset(Dataset[FIPSCertificate, FIPSAuxiliaryDatasets], ComplexSerial
df = pd.DataFrame([x.pandas_tuple for x in self.certs.values()], columns=FIPSCertificate.pandas_columns)
df = df.set_index("dgst")
- df.date_validation = pd.to_datetime(df.date_validation, infer_datetime_format=True, errors="coerce")
- df.date_sunset = pd.to_datetime(df.date_sunset, infer_datetime_format=True, errors="coerce")
+ df.date_validation = pd.to_datetime(df.date_validation, errors="coerce")
+ df.date_sunset = pd.to_datetime(df.date_sunset, errors="coerce")
# Manually delete one certificate with bad embodiment (seems to have many blank fields)
df = df.loc[~(df.embodiment == "*")]
diff --git a/src/sec_certs/model/cc_matching.py b/src/sec_certs/model/cc_matching.py
index 4e6f5735..21d94b74 100644
--- a/src/sec_certs/model/cc_matching.py
+++ b/src/sec_certs/model/cc_matching.py
@@ -75,10 +75,10 @@ class CCSchemeMatcher(AbstractMatcher[CCCertificate]):
if self._product == cert.name and self._vendor == cert.manufacturer:
return 99
# If we match the report hash, return early.
- if cert.state.report_pdf_hash == self._report_hash and self._report_hash is not None:
+ if cert.state.report.pdf_hash == self._report_hash and self._report_hash is not None:
return 95
# If we match the target hash, return early.
- if cert.state.st_pdf_hash == self._target_hash and self._target_hash is not None:
+ if cert.state.st.pdf_hash == self._target_hash and self._target_hash is not None:
return 93
# Fuzzy match at the end with some penalization.
diff --git a/src/sec_certs/model/references_nlp/segment_extractor.py b/src/sec_certs/model/references_nlp/segment_extractor.py
index d63208ec..f77c8f5b 100644
--- a/src/sec_certs/model/references_nlp/segment_extractor.py
+++ b/src/sec_certs/model/references_nlp/segment_extractor.py
@@ -173,9 +173,9 @@ class ReferenceSegmentExtractor:
- Loads manually annotated samples
- Combines all of that into single dataframe
"""
- target_certs = [x for x in certs if x.heuristics.st_references.directly_referencing and x.state.st_txt_path]
+ target_certs = [x for x in certs if x.heuristics.st_references.directly_referencing and x.state.st.txt_path]
report_certs = [
- x for x in certs if x.heuristics.report_references.directly_referencing and x.state.report_txt_path
+ x for x in certs if x.heuristics.report_references.directly_referencing and x.state.report.txt_path
]
df_targets = self._build_df(target_certs, "target")
df_reports = self._build_df(report_certs, "report")
@@ -217,8 +217,8 @@ class ReferenceSegmentExtractor:
for key, val in actual_references.items()
]
- (certs[0].state.report_txt_path.parent.parent / "txt_processed").mkdir(exist_ok=True, parents=True)
- (certs[0].state.st_txt_path.parent.parent / "txt_processed").mkdir(exist_ok=True, parents=True)
+ (certs[0].state.report.txt_path.parent.parent / "txt_processed").mkdir(exist_ok=True, parents=True)
+ (certs[0].state.st.txt_path.parent.parent / "txt_processed").mkdir(exist_ok=True, parents=True)
return list(itertools.chain.from_iterable(get_cert_records(cert, source) for cert in certs))
def _build_df(self, certs: list[CCCertificate], source: Literal["target", "report"]) -> pd.DataFrame:
diff --git a/src/sec_certs/rules.yaml b/src/sec_certs/rules.yaml
index 8973b015..43a309e1 100644
--- a/src/sec_certs/rules.yaml
+++ b/src/sec_certs/rules.yaml
@@ -5,75 +5,159 @@
#####
cc_cert_id:
DE:
- - "BSI-DSZ-CC-[0-9]+?-[0-9]+"
- - "BSI-DSZ-CC-[0-9]+?-(?:V|v)[0-9]+-[0-9]+"
- - "BSI-DSZ-CC-[0-9]+?-(?:V|v)[0-9]+"
- - "BSI-DSZ-CC-[0-9]+-(?:V|v)[0-9]+(-[0-9][0-9][0-9][0-9])*" # German BSI (number + version + year or without year)
- - "BSI-DSZ-CC-[0-9]+-[0-9][0-9][0-9][0-9]" # German BSI (number + year, no version)
- - "BSI-DSZ-CC-[0-9]+-(?:V|v)[0-9]+(?!-)" # German BSI (number + version but no year => no - after version)
- # - "BSI-DSZ-CC-[0-9]+" # Maybe?
+ - "BSI-DSZ-CC-(?:(?P<s>S)-)?(?P<counter>[0-9]{3,5})-?(?:(?P<version>[vV][0-9])-)?(?P<year>[0-9]{4})?(?:-(?P<doc>(?:RA|MA)(?:-[0-9]+)?))?"
+ # Examples:
+ # BSI-DSZ-CC-1004
+ # BSI-DSZ-CC-0973-2016
+ # BSI-DSZ-CC-0831-V4-2021
+ # BSI-DSZ-CC-0837-V2-2014-MA-01
+ # BSI-DSZ-CC-S-0192-2021
FR:
- - "ANSS[Ii](?:-|-CC-|-CC )[0-9]{4}/[0-9]+(v[1-9])?" # French
- - "ANSS[Ii]-CC[ -][0-9]{4}[/-_][0-9][0-9]+(?!-M|-S|-R)" # French (/two or more digits then NOT -M or -S)
- - "ANSS[Ii]-CC[ -][0-9]{4}[/-_][0-9]+(?:v[0-9])?[_/-][MSR][0-9]+" # French, maintenance or surveillance report (ANSSI-CC-2014_46_M01)
- # 'ANSSI-CC-CER-F-.+?', # French
- - "DCSS[Ii]-[0-9]+/[0-9]+" # French (DCSSI-2009/07)
- - "Certification Report [0-9]+/[0-9]+" # French or Australia! Solved because we limit ourselves to scheme when doing heuristics.
- - "Rapport de certification [0-9]+/[0-9]+" # French
+ - "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]))?"
+ # Examples:
+ # DCSSI-2009/07
+ # ANSSI-CC 2001/02-R01
+ # Rapport de certification 2001/02v2
+ # Certification Report 2003/20
NL:
- - "NSCIB-CC-[0-9][0-9][0-9][0-9].+?" # Netherlands
- - "NSCIB-CC-[0-9][0-9][0-9][0-9][0-9]*-CR" # Netherlands
- - "NSCIB-CC-[0-9][0-9]-[0-9]+?-CR[0-9]+?" # Netherlands
- - "NSCIB-CC-[0-9][0-9]-[0-9]+(-CR[0-9]+)*" # Netherlands (old number NSCIB-CC-05-6609 or NSCIB-CC-05-6609-CR)
- - "NSCIB-CC-[0-9]+-CR[0-9]*" # Netherlands (new number NSCIB-CC-111441-CR NSCIB-CC-111441-CR1)
- - "NSCIB-CC-[0-9]+-MA[0-9]*" # Netherlands (new number NSCIB-CC-222073-MA NSCIB-CC-200716-MA2)
- - "NSCIB-CC-[0-9][0-9]-[0-9]+" # Netherlands (old number NSCIB-CC-05-6609)
- - "NSCIB-CC-[0-9][0-9]-[0-9]+-CR[0-9]+" # Netherlands (NSCIB-CC-year2digits-number-CR)
+ - "(?:NSCIB-|CC-|NSCIB-CC-)(?P<core>((?P<year>[0-9]{2})-)?(?:-?[0-9]+)+)(?:-?(?P<doc>(?:CR|MA|MR)[0-9]*))?"
+ # Examples:
+ # NSCIB-CC-22-0428888-CR2 (with year=22 and CR2)
+ # NSCIB-CC-228723-CR (no year)
+ # CC-16-31801-CR4 (no NSCIB)
+ # NSCIB-CC-98209 (no year, no CR)
"NO":
- - "SERTIT-[0-9]+" # Norway
+ - "SERTIT-(?P<counter>[0-9]+)"
+ # Examples:
+ # SERTIT-101
US:
- - "CCEVS-VR-(?:CC-|VID)?[0-9]+-[0-9]+[a-z]?(?:-[0-9]+)?" # US NSA (CCEVS-VR-10884-2018 CCEVS-VR-VID10877-2018)
+ - "CCEVS-VR-(?:(?P<cc>CC)-)?(?:(?P<VID>VID)-?)?(?P<year>[0-9]{2})-(?P<counter>[0-9]+)"
+ - "CCEVS-VR-(?:(?P<cc>CC)-)?(?:(?P<VID>VID)-?)?(?P<counter>[0-9]{4,5})(?:-(?P<year>[0-9]{4}))?"
+ # Examples:
+ # CCEVS-VR-VID10015-2008
+ # CCEVS-VR-10880-2018
+ # CCEVS-VR-04-0082
+ # CCEVS-VR-VID10318
CA:
- # '[0-9][0-9\-]+?-CR', # Canada
- - "[0-9][0-9][0-9]-[347]-[0-9][0-9][0-9]?(?:-CR|P)?" # Canada xxx-{347}-xxx (383-4-438, 383-4-82-CR, 383-4-422P)
- - "[0-9][0-9][0-9][ -](?:EWA|LSS|CCS)(?:[ -]20[0-9][0-9])?" # Canada (522-EWA-2020, 524 LSS 2020, 503-LSS)
- - "[0-9][0-9][0-9](?:%20|-)(?:EWA|LSS|CCS)(?:%20|-)(?:20[0-9][0-9]%20|)CR%20v[0-9]\\.[0-9]" # Canada filename with space (518-LSS%20CR%20v1.0)
+ - "(?P<number1>383)[ -](?P<digit>[0-9])[ -](?P<number2>[0-9]+)(?:-CR|P)?"
+ - "(?P<number>[0-9]+)[ -](?P<lab>EWA|LSS|CCS)(?:[ -](?P<year>[0-9]+))?"
+ # Examples:
+ # 383-4-123-CR
+ # 383-4-123P
+ # 522 EWA 2020
+ # Filename rule:
+ #- "[0-9][0-9][0-9](?:%20|-)(?:EWA|LSS|CCS)(?:%20|-)(?:20[0-9][0-9]%20|)CR%20v[0-9]\\.[0-9]" # Canada filename with space (518-LSS%20CR%20v1.0)
UK:
- - "CRP[0-9]+[A-Z]?" # UK CESG
- - "CERTIFICATION REPORT No. P[0-9]+[A-Z]?" # UK CESG
+ - "CRP(?P<counter>[0-9]+[A-Z]?)"
+ - "CERTIFICATION REPORT No. P(?P<counter>[0-9]+[A-Z]?)"
+ # Examples:
+ # CRP208
+ # CERTIFICATION REPORT No. P123A
ES:
- - "20[0-9][0-9][-‐][0-9]+[-‐]INF[-‐][0-9]+([-‐]?[ -‐](?:V|v)[0-9]+)?" # Spain ("2006-4-INF-98 v2" or "2006-4-INF-98-v2" or "2020-34-INF-3784- v1")
+ - "(?P<year>[0-9]{4})[-‐](?P<project>[0-9]+)[-‐]INF[-‐](?P<counter>[0-9]+)[ -‐]{1,2}[vV](?P<version>[0-9])"
+ # Examples:
+ # 2006-4-INF-98 v2
+ # 2020-34-INF-3784- v1
+ # 2019-20-INF-3379-v1
KR:
- # Korea
+ - "KECS[-‐](?P<word>ISIS|NISS|CISS)[-‐](?P<counter>[0-9]{2,4})[-‐](?P<year>[0-9]{4})"
# XXX: Do not use KECS-CR as those refer to the certificate report and do not represent the certificate id.
- # - "KECS[-‐]CR[-‐][0-9]+[-‐][0-9]+" # Korea KECS-CR-20-61
- - "KECS[-‐](?:ISIS|NISS|CISS)[-‐][0-9]+[-‐][0-9]{4}" # Korea KECS-ISIS-1234-2011
+ # Examples:
+ # KECS-ISIS-0579-2015
+ # KECS-NISS-0792-2017
+ # KECS-CISS-1210-2023
JP:
- - "(?:CRP|ACR)-C[0-9]+-[0-9]+" # Japan (CRP-C0595-01 ACR-C0417-03)
- - "JISEC-CC-CRP-C[0-9]+-[0-9]+-[0-9]+" # Japan (JISEC-CC-CRP-C0689-01-2020)
- - "Certification No. [cC][0-9]+" # Japan (Certification No. C0090)
+ - "(?:CRP|ACR)-C(?P<counter>[0-9]+)-(?P<digit>[0-9]+)"
+ - "JISEC-CC-CRP-C(?P<counter>[0-9]+)(?:-(?P<digit>[0-9]{2}))?(?:-(?P<year>[0-9]{4}))?"
+ - "Certification No. [cC](?P<counter>[0-9]+)"
+ # Examples:
+ # CRP-C0595-01
+ # JISEC-CC-CRP-C0689-01-2020
+ # Certification No. C0090
MY:
- - "ISCB-[0-9]+-(?:RPT|FRM)-[CM][0-9]+[A-Z]?-(?:CR|AMR)(?:-[0-9])?-[vV][0-9](?:\\.[0-9])?[a-z]?" # Malaysia (ISCB-3-RPT-C092-CR-v1, ISCB-3-RPT-C068-CR-1-v1)
+ - "ISCB-(?P<digit>[0-9])-RPT-C(?P<counter>[0-9]{3})-CR(?:-[0-9])?-(?P<version>[vV][0-9][a-z]?)"
+ # Examples:
+ # ISCB-3-RPT-C068-CR-1-v1
+ # ISCB-5-RPT-C075-CR-v2
+ # ISCB-5-RPT-C046-CR-V1a
IT:
- - "OCSI/CERT/.+?" # Italy
- - "OCSI/CERT/.+?/20[0-9]+(?:\\w|/RC)" # Italy (OCSI/CERT/ATS/01/2018/RC)
+ - "OCSI/CERT/(?:(?P<lab>[A-Z]{3})/)?(?P<counter>[0-9]{2,3})/(?P<year>[0-9]{4})/RC"
+ # Examples:
+ # OCSI/CERT/SYS/04/2018/RC
+ # OCSI/CERT/CCL/10/2022/RC
+ # OCSI/CERT/TEC/09/2017/RC
+ # OCSI/CERT/ATS/06/2020/RC
TR:
- - "[0-9\\.]+?/TSE-CCCS-[0-9]+" # Turkish CCCS (21.0.0sc/TSE-CCCS-75)
- - "(?:[0-9]{1,2}\\.){2}[0-9]{1,2}/[0-9]{1,4}-[0-9]{3}" # 21.0.01/13-028
+ - "(?P<prefix>[0-9\\.]+)/TSE-CCCS-(?P<number>[0-9]+)"
+ # XXX: The report numbers are like "21.0.01/13-028"
+ # Examples:
+ # 21.0.03.0.00.00/TSE-CCCS-85
+ # 21.0.03/TSE-CCCS-33
SE:
- - "CSEC ?[0-9]{6,7}" # Sweden (CSEC2019015)
+ - "CSEC ?(?P<year>[0-9]{4})(?P<counter>[0-9]{2,3})"
+ # Examples:
+ # CSEC2019015
+ # CSEC 2019012
IN:
- # India (IC3S/DEL01/VALIANT/EAL1/0317/0007/CR STQC/CC/14-15/12/ETR/0017 IC3S/MUM01/CISCO/cPP/0119/0016/CR)
- # will miss STQC/CC/14-15/12/ETR/0017
- - "(?:IC3S|STQC/CC)/[^ ]+? ?/CR"
+ - "IC3S/(?P<lab>[A-Z]+[0-9]+)/(?P<vendor>[a-zA-Z_]+)/(?P<level>[a-zA-Z0-9]+)/(?P<number1>[0-9]+)/(?P<number2>[0-9]+) ?(?:/CR)?"
+ # XXX: The cert IDs are often present only in the certificate and not in the report.
+ # The report often only has the report id, of the format "STQC/CC/1617/18/CR"
+ # Examples:
+ # IC3S/BG01/HALTDOS/EAL2/0317/0008/CR
+ # IC3S/KOL01/ADVA/EAL2/0520/0021/CR
+ # IC3S/MUM01/Symantec/NDcPP/0722/0032/CR
SG:
- - "CSA_CC_[0-9]+" # Singapore (CSA_CC_19001)
+ - "CSA_CC_(?P<year>[0-9]{2})(?P<counter>[0-9]{3})"
+ # Examples:
+ # CSA_CC_19001
AU:
- # Australia (EFS-T048 ETR 1.0, EFS-T056-ETR 1.0, DXC-EFC-T092-ETR 1.0)
+ - "(?:Certificate Number:|Certification Report) (?P<year>[0-9]{2,4})/(?P<counter>[0-9]+)"
# XXX: Do not use Australian ETR numbers, they are not certificate id.
- # - "(?:EFS|EFT|DXC-EFC)-T[0-9]+(?: |-)ETR [0-9]+.[0-9]+"
- - "Certificate Number: [0-9]{1,4}/[0-9]{1,4}"
- - "Certification Report [0-9]+/[0-9]+"
+ # Examples:
+ # Certification Report 2007/06
+ # Certificate Number: 2010/67
+ # Certificate Number: 37/2006 !mistake
+ # Certification Report 97/76 !short year
+
+#####
+# Common Criteria certificate IDs as they appear in report filenames, grouped by scheme (Alpha-2 ISO country code).
+#####
+cc_filename_cert_id:
+ DE:
+ #- "(?P<counter>[0-9]{3,5})(?:(?P<version>[vV][0-9]))?a?(?:_pdf)?"
+ - "(?P<year>[0-9]{4})(?P<month>[0-9]{2})(?P<day>[0-9]{2})_(?P<counter>[0-9]{3,5})(?:(?P<version>[vV][0-9]))?a?(?:_pdf)?"
+ FR:
+ - "(?P<year>[0-9]{4})[_-](?P<counter>[0-9]{2})([vV](?P<version>[0-9]))?"
+ #- "(?P<year>[0-9]{2})(?P<counter>[0-9]{2})([vV](?P<version>[0-9]))?"
+ NL:
+ - "(?:NSCIB-|CC-|NSCIB-CC-)(?P<core>((?P<year>[0-9]{2})-)?(?:-?[0-9]+)+)(?:-?(?P<doc>(?:CR|MA|MR)[0-9]*))?"
+ "NO":
+ - "SERTIT-(?P<counter>[0-9]+)"
+ US:
+ CA:
+ - "(?P<number1>[0-9]+)[ -](?P<digit>[0-9])[ -](?P<number2>[0-9]+)(?:-CR|P)?"
+ - "(?P<number>[0-9]+)[ -](?P<lab>EWA|LSS|CCS)(?:[ -](?P<year>[0-9]+))?"
+ UK:
+ - "CRP(?P<counter>[0-9]+[A-Z]?)"
+ ES:
+ - "(?P<year>[0-9]{4})[-‐](?P<project>[0-9]+)[-‐]INF[-‐](?P<counter>[0-9]+)[ -‐_]{1,2}[vV](?P<version>[0-9])"
+ KR:
+ - "(?P<word>ISIS|NISS|CISS)[-‐](?P<counter>[0-9]{2,4})[-‐](?P<year>[0-9]{4})"
+ JP:
+ - "[cC](?P<counter>[0-9]+)"
+ MY:
+ - "ISCB-(?P<digit>[0-9])-RPT-C(?P<counter>[0-9]{3})-CR(?:-[0-9])?-(?P<version>[vV][0-9][a-z]?)"
+ IT:
+ TR:
+ SE:
+ - "CR(?P<year>[0-9]{4})(?P<counter>[0-9]{2,3})"
+ IN:
+ SG:
+ AU:
+ - "(?P<year>[0-9]{2,4})_(?P<counter>[0-9]+)"
#####
# Common Criteria protection profile IDs, grouped by certification body (e.g. BSI)
@@ -1083,39 +1167,39 @@ fips_security_level:
#####
fips_certlike:
Certlike:
- # --- HMAC(-SHA)(-1) - (bits) (method) ((hardware/firmware cert) #id) ---
- # + added (and #id) everywhere
+ # --- HMAC(-SHA)(-1) - (bits) (method) ((hardware/firmware cert) #id) ---
+ # + added (and #id) everywhere
- "HMAC(?:[- –]*SHA)?(?:[- –]*1)?[– -]*((?:;|\\/|160|224|256|384|512)?(?:;|\\/| |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[, ]*?\\(?(?: |hardware|firmware)*?[\\s(\\[]*?(?:#|cert\\.?|Cert\\.?|Certificate|sample)?[\\s#]*?)?[\\s#]*?(\\d{1,4})(?:[\\s#]*and[\\s#]*\\d+)?"
- # --- same as above, without hw or fw ---
+ # --- same as above, without hw or fw ---
- "HMAC(?:-SHA)?(?:-1)?[ -]*((?:;|\\/|160|224|256|384|512)?(?:;|\\/| |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[, ]*?(?:#|cert\\.?|sample|Cert\\.?|Certificate)?[\\s#]*?)?[\\s#]*?(\\d{1,4})"
- # --- SHS/A - (bits) (method) ((cert #) numbers) ---
+ # --- SHS/A - (bits) (method) ((cert #) numbers) ---
- "SH[SA][-– 123]*(?:;|\\/|160|224|256|384|512)?(?:[\\s(\\[]*?(?:KAT|[Bb]yte [Oo]riented)*?[\\s,]*?[\\s(\\[]*?(?:#|cert\\.?|sample|Cert\\.?|Certificate)?[\\s#]*?)?[\\s#]*?(\\d{1,4})(?:\\)?\\[#?\\d+\\])?(?:[\\s#]*?and[\\s#]*?\\d+)?"
- # --- RSA (bits) (method) ((cert #)) ---
+ # --- RSA (bits) (method) ((cert #)) ---
- "RSA(?:[-– ]*(?:;|\\/|512|768|1024|1280|1536|2048|3072|4096|8192)\\s\\(\\[]*?(?:(?:;|\\/|KAT|Verify|PSS|\\s)*?)?[\\s,]*?[\\s(\\[]*?(?:#|cert\\.?|sample|Cert\\.?|Certificate)?[\\s#]*?)?[\\s#]*?(\\d{1,4})"
- # --- RSA (SSA) (PKCS) (version) (#) ---
+ # --- RSA (SSA) (PKCS) (version) (#) ---
- "(?:RSA)?[-– ]?(?:SSA)?[- ]?PKCS\\s?#?\\d(?:-[Vv]1_5| [Vv]1[-_]5)?[\\s#]*?(\\d{1,4})?"
- # --- AES (bits) (method) ((cert #)) ---
+ # --- AES (bits) (method) ((cert #)) ---
- "AES[-– ]*((?: |;|\\/|bit|key|128|192|256|CBC)*(?: |\\/|;|[Dd]ecrypt|[Ee]ncrypt|KAT|CMAC|CTR|GCM|IV|CBC)*?[,\\s(\\[]*?(?:#|cert\\.?|sample|Cert\\.?|Certificate)?[\\s#]*?)?[\\s#]*?(\\d{1,4})(?:\\)?[\\s#]*?\\[#?\\d+\\])?(?:[\\s#]*?and[\\s#]*?(\\d+))?"
- # --- Diffie Helman (CVL) ((cert #)) ---
+ # --- Diffie Helman (CVL) ((cert #)) ---
- "Diffie[-– ]*Hellman[,\\s(\\[]*?(?:CVL|\\s)*?(?:#|cert\\.?|sample|Cert\\.?|Certificate)?[\\s#]*?[\\s#]*?(\\d{1,4})"
- # --- DRBG (bits) (method) (cert #) ---
+ # --- DRBG (bits) (method) (cert #) ---
- "DRBG[ –-]*((?:;|\\/|160|224|256|384|512)?(?:;|\\/| |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\\s(\\[]*?(?:#|cert\\.?|sample|Cert\\.?|Certificate)?[\\s#]*?)?[\\s#]*?(\\d{1,4})"
- # --- DES (bits) (method) (cert #)
+ # --- DES (bits) (method) (cert #)
- "DES[ –-]*((?:;|\\/|160|224|256|384|512)?(?:;|\\/| |[Dd]ecrypt|[Ee]ncrypt|KAT|CBC|(?:\\d(?: and \\d)? keying options?))*?[,\\s(\\[]*?(?:#|cert\\.?|sample|Cert\\.?|Certificate)*?[\\s#]*?)?[\\s#]*?(\\d{1,4})(?:[\\s#]*?and[\\s#]*?(\\d+))?"
- # --- DSA (bits) (method) (cert #)
+ # --- DSA (bits) (method) (cert #)
- "DSA[ –-]*((?:;|\\/|160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\\s(\\[]*?(?:#|cert\\.?|sample|Cert\\.?|Certificate)?[\\s#]*?)?[\\s#]*?(\\d{1,4})"
- # --- platforms (#)+ - this is used in modification history ---
+ # --- platforms (#)+ - this is used in modification history ---
- "[Pp]latforms? #\\d+(?:#\\d+|,| |-|and)*[^\\n]*"
- # --- CVL (#) ---
+ # --- CVL (#) ---
- "CVL[\\s#]*?(\\d{1,4})"
- # --- PAA (#) ---
+ # --- PAA (#) ---
- "PAA[: #]*?\\d{1,4}"
- # --- (#) Type ---
+ # --- (#) Type ---
- "(?:#|cert\\.?|sample|Cert\\.?|Certificate)[\\s#]*?(\\d+)?\\s*?(?:AES|SHS|SHA|RSA|HMAC|Diffie-Hellman|DRBG|DES|CVL)"
- # --- PKCS (#) ---
+ # --- PKCS (#) ---
- "PKCS[\\s]?#?\\d+"
- "PKSC[\\s]?#?\\d+" # typo, #625
- # --- # C and # A (just in case) ---
+ # --- # C and # A (just in case) ---
- "#\\s+?[Cc]\\d+"
- "#\\s+?[Aa]\\d+"
diff --git a/src/sec_certs/sample/cc.py b/src/sec_certs/sample/cc.py
index 32c875c8..28099f58 100644
--- a/src/sec_certs/sample/cc.py
+++ b/src/sec_certs/sample/cc.py
@@ -5,7 +5,6 @@ import re
from collections import Counter, defaultdict
from dataclasses import dataclass, field
from datetime import date, datetime
-from enum import Enum
from pathlib import Path
from typing import Any, ClassVar
from urllib.parse import unquote_plus, urlparse
@@ -19,7 +18,7 @@ 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.sample.cc_certificate_id import canonicalize
+from sec_certs.sample.cc_certificate_id import canonicalize, schemes
from sec_certs.sample.certificate import Certificate, References, logger
from sec_certs.sample.certificate import Heuristics as BaseHeuristics
from sec_certs.sample.certificate import PdfData as BasePdfData
@@ -28,20 +27,7 @@ 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.extract import normalize_match_string
-
-HEADERS = {
- "anssi": sec_certs.utils.extract.search_only_headers_anssi,
- "bsi": sec_certs.utils.extract.search_only_headers_bsi,
- "nscib": sec_certs.utils.extract.search_only_headers_nscib,
- "niap": sec_certs.utils.extract.search_only_headers_niap,
- "canada": sec_certs.utils.extract.search_only_headers_canada,
-}
-
-
-class ReferenceType(Enum):
- DIRECT = "direct"
- INDIRECT = "indirect"
+from sec_certs.utils.extract import normalize_match_string, scheme_frontpage_functions
class CCCertificate(
@@ -95,141 +81,87 @@ class CCCertificate(
def __lt__(self, other):
return self.maintenance_date < other.maintenance_date
- @dataclass(init=False)
- class InternalState(ComplexSerializableType):
- """
- Holds internal state of the certificate, whether downloads and converts of individual components succeeded. Also
- holds information about errors and paths to the files.
- """
-
- st_download_ok: bool # Whether target download went OK
- report_download_ok: bool # Whether report download went OK
- st_convert_garbage: bool # Whether initial target conversion resulted in garbage
- report_convert_garbage: bool # Whether initial report conversion resulted in garbage
- st_convert_ok: bool # Whether overall target conversion went OK (either pdftotext or via OCR)
- report_convert_ok: bool # Whether overall report conversion went OK (either pdftotext or via OCR)
- st_extract_ok: bool # Whether target extraction went OK
- report_extract_ok: bool # Whether report extraction went OK
+ @dataclass
+ class DocumentState(ComplexSerializableType):
+ download_ok: bool = False # Whether download went OK
+ convert_garbage: bool = False # Whether initial conversion resulted in garbage
+ convert_ok: bool = False # Whether overall conversion went OK (either pdftotext or via OCR)
+ extract_ok: bool = False # Whether extraction went OK
- st_pdf_hash: str | None
- report_pdf_hash: str | None
- st_txt_hash: str | None
- report_txt_hash: str | None
+ pdf_hash: str | None = None
+ txt_hash: str | None = None
- _st_pdf_path: Path | None = None
- _report_pdf_path: Path | None = None
- _st_txt_path: Path | None = None
- _report_txt_path: Path | None = None
+ _pdf_path: Path | None = None
+ _txt_path: Path | None = None
- def __init__(
- self,
- st_download_ok: bool = False,
- report_download_ok: bool = False,
- st_convert_garbage: bool = False,
- report_convert_garbage: bool = False,
- st_convert_ok: bool = False,
- report_convert_ok: bool = False,
- st_extract_ok: bool = False,
- report_extract_ok: bool = False,
- st_pdf_hash: str | None = None,
- report_pdf_hash: str | None = None,
- st_txt_hash: str | None = None,
- report_txt_hash: str | None = None,
- ):
- super().__init__()
- self.st_download_ok = st_download_ok
- self.report_download_ok = report_download_ok
- self.st_convert_garbage = st_convert_garbage
- self.report_convert_garbage = report_convert_garbage
- self.st_convert_ok = st_convert_ok
- self.report_convert_ok = report_convert_ok
- self.st_extract_ok = st_extract_ok
- self.report_extract_ok = report_extract_ok
- self.st_pdf_hash = st_pdf_hash
- self.report_pdf_hash = report_pdf_hash
- self.st_txt_hash = st_txt_hash
- self.report_txt_hash = report_txt_hash
+ def is_ok_to_download(self, fresh: bool = True) -> bool:
+ return True if fresh else not self.download_ok
- @property
- def st_pdf_path(self) -> Path:
- if not self._st_pdf_path:
- raise ValueError(f"st_pdf_path not set on {type(self)}")
- return self._st_pdf_path
-
- @st_pdf_path.setter
- def st_pdf_path(self, pth: str | Path | None) -> None:
- self._st_pdf_path = Path(pth) if pth else None
-
- @property
- def report_pdf_path(self) -> Path:
- if not self._report_pdf_path:
- raise ValueError(f"report_pdf_path not set on {type(self)}")
- return self._report_pdf_path
+ def is_ok_to_convert(self, fresh: bool = True) -> bool:
+ return self.download_ok if fresh else self.download_ok and not self.convert_ok
- @report_pdf_path.setter
- def report_pdf_path(self, pth: str | Path | None) -> None:
- self._report_pdf_path = Path(pth) if pth else None
+ def is_ok_to_analyze(self, fresh: bool = True) -> bool:
+ if fresh:
+ return self.download_ok and self.convert_ok
+ else:
+ return self.download_ok and self.convert_ok and not self.extract_ok
@property
- def st_txt_path(self) -> Path:
- if not self._st_txt_path:
- raise ValueError(f"st_txt_path not set on {type(self)}")
- return self._st_txt_path
+ def pdf_path(self) -> Path:
+ if not self._pdf_path:
+ raise ValueError(f"pdf_path not set on {type(self)}")
+ return self._pdf_path
- @st_txt_path.setter
- def st_txt_path(self, pth: str | Path | None) -> None:
- self._st_txt_path = Path(pth) if pth else None
+ @pdf_path.setter
+ def pdf_path(self, pth: str | Path | None) -> None:
+ self._pdf_path = Path(pth) if pth else None
@property
- def report_txt_path(self) -> Path:
- if not self._report_txt_path:
- raise ValueError(f"report_txt_path not set on {type(self)}")
- return self._report_txt_path
+ def txt_path(self) -> Path:
+ if not self._txt_path:
+ raise ValueError(f"txt_path not set on {type(self)}")
+ return self._txt_path
- @report_txt_path.setter
- def report_txt_path(self, pth: str | Path | None) -> None:
- self._report_txt_path = Path(pth) if pth else None
+ @txt_path.setter
+ def txt_path(self, pth: str | Path | None) -> None:
+ self._txt_path = Path(pth) if pth else None
@property
def serialized_attributes(self) -> list[str]:
return [
- "st_download_ok",
- "report_download_ok",
- "st_convert_garbage",
- "report_convert_garbage",
- "st_convert_ok",
- "report_convert_ok",
- "st_extract_ok",
- "report_extract_ok",
- "st_pdf_hash",
- "report_pdf_hash",
- "st_txt_hash",
- "report_txt_hash",
+ "download_ok",
+ "convert_garbage",
+ "convert_ok",
+ "extract_ok",
+ "pdf_hash",
+ "txt_hash",
]
- def report_is_ok_to_download(self, fresh: bool = True) -> bool:
- return True if fresh else not self.report_download_ok
-
- def st_is_ok_to_download(self, fresh: bool = True) -> bool:
- return True if fresh else not self.st_download_ok
-
- def report_is_ok_to_convert(self, fresh: bool = True) -> bool:
- return self.report_download_ok if fresh else self.report_download_ok and not self.report_convert_ok
+ @dataclass(init=False)
+ class InternalState(ComplexSerializableType):
+ """
+ Holds internal state of the certificate, whether downloads and converts of individual components succeeded. Also
+ holds information about errors and paths to the files.
+ """
- def st_is_ok_to_convert(self, fresh: bool = True) -> bool:
- return self.st_download_ok if fresh else self.st_download_ok and not self.st_convert_ok
+ report: CCCertificate.DocumentState
+ st: CCCertificate.DocumentState
+ cert: CCCertificate.DocumentState
- def report_is_ok_to_analyze(self, fresh: bool = True) -> bool:
- if fresh is True:
- return self.report_download_ok and self.report_convert_ok
- else:
- return self.report_download_ok and self.report_convert_ok and not self.report_extract_ok
+ def __init__(
+ self,
+ report: CCCertificate.DocumentState | None = None,
+ st: CCCertificate.DocumentState | None = None,
+ cert: CCCertificate.DocumentState | None = None,
+ ):
+ super().__init__()
+ self.report = report if report is not None else CCCertificate.DocumentState()
+ self.st = st if st is not None else CCCertificate.DocumentState()
+ self.cert = cert if cert is not None else CCCertificate.DocumentState()
- def st_is_ok_to_analyze(self, fresh: bool = True) -> bool:
- if fresh is True:
- return self.st_download_ok and self.st_convert_ok
- else:
- return self.st_download_ok and self.st_convert_ok and not self.st_extract_ok
+ @property
+ def serialized_attributes(self) -> list[str]:
+ return ["report", "st", "cert"]
@dataclass
class PdfData(BasePdfData, ComplexSerializableType):
@@ -239,139 +171,108 @@ class CCCertificate(
report_metadata: dict[str, Any] | None = field(default=None)
st_metadata: dict[str, Any] | None = field(default=None)
+ cert_metadata: dict[str, Any] | None = field(default=None)
report_frontpage: dict[str, dict[str, Any]] | None = field(default=None)
- st_frontpage: dict[str, dict[str, Any]] | None = field(default=None)
+ st_frontpage: dict[str, dict[str, Any]] | None = field(
+ default=None
+ ) # TODO: Unused, we have no frontpage matching for targets
+ cert_frontpage: dict[str, dict[str, Any]] | None = field(
+ default=None
+ ) # TODO: Unused, we have no frontpage matching for certs
report_keywords: dict[str, Any] | None = field(default=None)
st_keywords: dict[str, Any] | None = field(default=None)
+ cert_keywords: dict[str, Any] | None = field(default=None)
report_filename: str | None = field(default=None)
st_filename: str | None = field(default=None)
+ cert_filename: str | None = field(default=None)
def __bool__(self) -> bool:
return any(x is not None for x in vars(self))
@property
- def bsi_data(self) -> dict[str, Any] | None:
- """
- Returns frontpage data related to BSI-provided information
- """
- return self.report_frontpage.get("bsi", None) if self.report_frontpage else None
-
- @property
- def niap_data(self) -> dict[str, Any] | None:
- """
- Returns frontpage data related to niap-provided information
- """
- return self.report_frontpage.get("niap", None) if self.report_frontpage else None
-
- @property
- def nscib_data(self) -> dict[str, Any] | None:
- """
- Returns frontpage data related to nscib-provided information
- """
- return self.report_frontpage.get("nscib", None) if self.report_frontpage else None
-
- @property
- def canada_data(self) -> dict[str, Any] | None:
- """
- Returns frontpage data related to canada-provided information
- """
- return self.report_frontpage.get("canada", None) if self.report_frontpage else None
-
- @property
- def anssi_data(self) -> dict[str, Any] | None:
- """
- Returns frontpage data related to ANSSI-provided information
- """
- return self.report_frontpage.get("anssi", None) if self.report_frontpage else None
-
- @property
def cert_lab(self) -> list[str] | None:
"""
Returns labs for which certificate data was parsed.
"""
+ if not self.report_frontpage:
+ return None
labs = [
data["cert_lab"].split(" ")[0].upper()
- for data in [self.bsi_data, self.anssi_data, self.niap_data, self.nscib_data, self.canada_data]
- if data
+ for scheme, data in self.report_frontpage.items()
+ if data and "cert_lab" in data
]
return labs if labs else None
- @property
- def bsi_cert_id(self) -> str | None:
- return self.bsi_data.get("cert_id", None) if self.bsi_data else None
-
- @property
- def niap_cert_id(self) -> str | None:
- return self.niap_data.get("cert_id", None) if self.niap_data else None
-
- @property
- def nscib_cert_id(self) -> str | None:
- return self.nscib_data.get("cert_id", None) if self.nscib_data else None
-
- @property
- def canada_cert_id(self) -> str | None:
- return self.canada_data.get("cert_id", None) if self.canada_data else None
-
- @property
- def anssi_cert_id(self) -> str | None:
- return self.anssi_data.get("cert_id", None) if self.anssi_data else None
-
def frontpage_cert_id(self, scheme: str) -> dict[str, float]:
"""
Get cert_id candidate from the frontpage of the report.
"""
- scheme_map = {
- "DE": self.bsi_cert_id,
- "US": self.niap_cert_id,
- "NL": self.nscib_cert_id,
- "CA": self.canada_cert_id,
- "FR": self.anssi_cert_id,
- }
- if scheme in scheme_map and (candidate := scheme_map[scheme]):
- return {candidate: 1.0}
- return {}
+ if not self.report_frontpage:
+ return {}
+ data = self.report_frontpage.get(scheme)
+ if not data:
+ return {}
+ cert_id = data.get("cert_id")
+ if not cert_id:
+ return {}
+ else:
+ return {cert_id: 1.0}
def filename_cert_id(self, scheme: str) -> dict[str, float]:
"""
- Get cert_id candidates from the matches in the report filename.
+ Get cert_id candidates from the matches in the report filename and cert filename.
"""
- if not self.report_filename:
- return {}
- scheme_rules = rules["cc_cert_id"][scheme]
- matches: Counter = Counter()
- for rule in scheme_rules:
- match = re.search(rule, self.report_filename)
- if match:
- cert_id = normalize_match_string(match.group())
- matches[cert_id] += 1
- if not matches:
+ scheme_filename_rules = rules["cc_filename_cert_id"][scheme]
+ if not scheme_filename_rules:
return {}
- total = max(matches.values())
- results = {}
- for candidate, count in matches.items():
- results[candidate] = count / total
+ scheme_meta = schemes[scheme]
+ results: dict[str, float] = {}
+ for fname in (self.report_filename, self.cert_filename):
+ if not fname:
+ continue
+
+ matches: Counter = Counter()
+ for rule in scheme_filename_rules:
+ match = re.search(rule, fname)
+ if match:
+ try:
+ meta = match.groupdict()
+ cert_id = scheme_meta(meta)
+ matches[cert_id] += 1
+ except Exception:
+ continue
+ if not matches:
+ continue
+ total = max(matches.values())
+
+ for candidate, count in matches.items():
+ results.setdefault(candidate, 0)
+ results[candidate] += count / total
# TODO count length in weight
return results
def keywords_cert_id(self, scheme: str) -> dict[str, float]:
"""
- Get cert_id candidates from the keywords matches in the report.
+ Get cert_id candidates from the keywords matches in the report and cert.
"""
- if not self.report_keywords:
- return {}
- cert_id_matches = self.report_keywords.get("cc_cert_id")
- if not cert_id_matches:
- return {}
+ results: dict[str, float] = {}
+ for keywords in (self.report_keywords, self.cert_keywords):
+ if not keywords:
+ continue
+ cert_id_matches = keywords.get("cc_cert_id")
+ if not cert_id_matches:
+ continue
- if scheme not in cert_id_matches:
- return {}
- matches: Counter = Counter(cert_id_matches[scheme])
- if not matches:
- return {}
- total = max(matches.values())
- results = {}
- for candidate, count in matches.items():
- results[candidate] = count / total
+ if scheme not in cert_id_matches:
+ continue
+ matches: Counter = Counter(cert_id_matches[scheme])
+ if not matches:
+ continue
+ total = max(matches.values())
+
+ for candidate, count in matches.items():
+ results.setdefault(candidate, 0)
+ results[candidate] += count / total
# TODO count length in weight
return results
@@ -381,22 +282,27 @@ class CCCertificate(
"""
scheme_rules = rules["cc_cert_id"][scheme]
fields = ("/Title", "/Subject")
- matches: Counter = Counter()
- for meta_field in fields:
- field_val = self.report_metadata.get(meta_field) if self.report_metadata else None
- if not field_val:
+ results: dict[str, float] = {}
+ for metadata in (self.report_metadata, self.cert_metadata):
+ if not metadata:
continue
- for rule in scheme_rules:
- match = re.search(rule, field_val)
- if match:
- cert_id = normalize_match_string(match.group())
- matches[cert_id] += 1
- if not matches:
- return {}
- total = max(matches.values())
- results = {}
- for candidate, count in matches.items():
- results[candidate] = count / total
+ matches: Counter = Counter()
+ for meta_field in fields:
+ field_val = metadata.get(meta_field)
+ if not field_val:
+ continue
+ for rule in scheme_rules:
+ match = re.search(rule, field_val)
+ if match:
+ cert_id = normalize_match_string(match.group())
+ matches[cert_id] += 1
+ if not matches:
+ continue
+ total = max(matches.values())
+
+ for candidate, count in matches.items():
+ results.setdefault(candidate, 0)
+ results[candidate] += count / total
# TODO count length in weight
return results
@@ -410,14 +316,28 @@ class CCCertificate(
candidates: dict[str, float] = defaultdict(lambda: 0.0)
# TODO: Add heuristic based on ordering of ids (and extracted year + increment)
# TODO: Add heuristic based on length
+ # TODO: Add heuristic based on id "richness", we want to prefer IDs that have more components.
+ # If we cannot canonicalize, just skip that ID.
for candidate, count in frontpage_id.items():
- candidates[canonicalize(candidate, scheme)] += count * 1.5
+ try:
+ candidates[canonicalize(candidate, scheme)] += count * 1.5
+ except Exception:
+ continue
for candidate, count in metadata_id.items():
- candidates[canonicalize(candidate, scheme)] += count * 1.2
+ try:
+ candidates[canonicalize(candidate, scheme)] += count * 1.2
+ except Exception:
+ continue
for candidate, count in keywords_id.items():
- candidates[canonicalize(candidate, scheme)] += count * 1.0
+ try:
+ candidates[canonicalize(candidate, scheme)] += count * 1.0
+ except Exception:
+ continue
for candidate, count in filename_id.items():
- candidates[canonicalize(candidate, scheme)] += count * 1.0
+ try:
+ candidates[canonicalize(candidate, scheme)] += count * 1.0
+ except Exception:
+ continue
return candidates
@dataclass
@@ -486,8 +406,8 @@ class CCCertificate(
security_level: str | set[str],
not_valid_before: date | None,
not_valid_after: date | None,
- report_link: str,
- st_link: str,
+ report_link: str | None,
+ st_link: str | None,
cert_link: str | None,
manufacturer_web: str | None,
protection_profiles: set[ProtectionProfile] | None,
@@ -694,13 +614,19 @@ class CCCertificate(
return extracted_date
@staticmethod
- def _html_row_get_report_st_links(cell: Tag) -> tuple[str, str]:
+ def _html_row_get_report_st_links(cell: Tag) -> tuple[str | None, str | None]:
links = cell.find_all("a")
- assert links[1].get("title").startswith("Certification Report")
- assert links[2].get("title").startswith("Security Target")
- report_link = CCCertificate.cc_url + links[1].get("href")
- security_target_link = CCCertificate.cc_url + links[2].get("href")
+ report_link: str | None = None
+ security_target_link: str | None = None
+ for link in links:
+ title = link.get("title")
+ if not title:
+ continue
+ if title.startswith("Certification Report"):
+ report_link = CCCertificate.cc_url + link.get("href")
+ elif title.startswith("Security Target"):
+ security_target_link = CCCertificate.cc_url + link.get("href")
return report_link, security_target_link
@@ -787,25 +713,33 @@ class CCCertificate(
self,
report_pdf_dir: str | Path | None,
st_pdf_dir: str | Path | None,
+ cert_pdf_dir: str | Path | None,
report_txt_dir: str | Path | None,
st_txt_dir: str | Path | None,
+ cert_txt_dir: str | Path | None,
) -> None:
"""
Sets paths to files given the requested directories
:param Optional[Union[str, Path]] report_pdf_dir: Directory where pdf reports shall be stored
:param Optional[Union[str, Path]] st_pdf_dir: Directory where pdf security targets shall be stored
+ :param Optional[Union[str, Path]] cert_pdf_dir: Directory where pdf certificates shall be stored
:param Optional[Union[str, Path]] report_txt_dir: Directory where txt reports shall be stored
:param Optional[Union[str, Path]] st_txt_dir: Directory where txt security targets shall be stored
+ :param Optional[Union[str, Path]] cert_txt_dir: Directory where txtcertificates shall be stored
"""
if report_pdf_dir:
- self.state.report_pdf_path = Path(report_pdf_dir) / (self.dgst + ".pdf")
+ self.state.report.pdf_path = Path(report_pdf_dir) / (self.dgst + ".pdf")
if st_pdf_dir:
- self.state.st_pdf_path = Path(st_pdf_dir) / (self.dgst + ".pdf")
+ self.state.st.pdf_path = Path(st_pdf_dir) / (self.dgst + ".pdf")
+ if cert_pdf_dir:
+ self.state.cert.pdf_path = Path(cert_pdf_dir) / (self.dgst + ".pdf")
if report_txt_dir:
- self.state.report_txt_path = Path(report_txt_dir) / (self.dgst + ".txt")
+ self.state.report.txt_path = Path(report_txt_dir) / (self.dgst + ".txt")
if st_txt_dir:
- self.state.st_txt_path = Path(st_txt_dir) / (self.dgst + ".txt")
+ self.state.st.txt_path = Path(st_txt_dir) / (self.dgst + ".txt")
+ if cert_txt_dir:
+ self.state.cert.txt_path = Path(cert_txt_dir) / (self.dgst + ".txt")
@staticmethod
def download_pdf_report(cert: CCCertificate) -> CCCertificate:
@@ -819,14 +753,14 @@ class CCCertificate(
if not cert.report_link:
exit_code = "No link"
else:
- exit_code = helpers.download_file(cert.report_link, cert.state.report_pdf_path)
+ exit_code = helpers.download_file(cert.report_link, cert.state.report.pdf_path)
if exit_code != requests.codes.ok:
error_msg = f"failed to download report from {cert.report_link}, code: {exit_code}"
logger.error(f"Cert dgst: {cert.dgst} " + error_msg)
- cert.state.report_download_ok = False
+ cert.state.report.download_ok = False
else:
- cert.state.report_download_ok = True
- cert.state.report_pdf_hash = helpers.get_sha256_filepath(cert.state.report_pdf_path)
+ cert.state.report.download_ok = True
+ cert.state.report.pdf_hash = helpers.get_sha256_filepath(cert.state.report.pdf_path)
cert.pdf_data.report_filename = unquote_plus(str(urlparse(cert.report_link).path).split("/")[-1])
return cert
@@ -839,39 +773,61 @@ class CCCertificate(
:return CCCertificate: returns the modified certificate with updated state
"""
exit_code: str | int = (
- helpers.download_file(cert.st_link, cert.state.st_pdf_path) if cert.st_link else "No link"
+ helpers.download_file(cert.st_link, cert.state.st.pdf_path) if cert.st_link else "No link"
)
if exit_code != requests.codes.ok:
error_msg = f"failed to download ST from {cert.st_link}, code: {exit_code}"
logger.error(f"Cert dgst: {cert.dgst} " + error_msg)
- cert.state.st_download_ok = False
+ cert.state.st.download_ok = False
else:
- cert.state.st_download_ok = True
- cert.state.st_pdf_hash = helpers.get_sha256_filepath(cert.state.st_pdf_path)
+ cert.state.st.download_ok = True
+ cert.state.st.pdf_hash = helpers.get_sha256_filepath(cert.state.st.pdf_path)
cert.pdf_data.st_filename = unquote_plus(str(urlparse(cert.st_link).path).split("/")[-1])
return cert
@staticmethod
+ def download_pdf_cert(cert: CCCertificate) -> CCCertificate:
+ """
+ Downloads pdf of the certificate. Staticmethod to allow for parallelization.
+
+ :param CCCertificate cert: cert to download the pdf of
+ :return CCCertificate: returns the modified certificate with updated state
+ """
+ exit_code: str | int = (
+ helpers.download_file(cert.cert_link, cert.state.cert.pdf_path) if cert.cert_link else "No link"
+ )
+
+ if exit_code != requests.codes.ok:
+ error_msg = f"failed to download certificate from {cert.cert_link}, code: {exit_code}"
+ logger.error(f"Cert dgst: {cert.dgst} " + error_msg)
+ cert.state.cert.download_ok = False
+ else:
+ cert.state.cert.download_ok = True
+ cert.state.cert.pdf_hash = helpers.get_sha256_filepath(cert.state.cert.pdf_path)
+ cert.pdf_data.cert_filename = unquote_plus(str(urlparse(cert.cert_link).path).split("/")[-1])
+ return cert
+
+ @staticmethod
def convert_report_pdf(cert: CCCertificate) -> CCCertificate:
"""
Converts the pdf certification report to txt, given the certificate. Staticmethod to allow for parallelization.
- :param CCCertificate cert: cert to download the pdf report for
+ :param CCCertificate cert: cert to convert the pdf report for
:return CCCertificate: the modified certificate with updated state
"""
ocr_done, ok_result = sec_certs.utils.pdf.convert_pdf_file(
- cert.state.report_pdf_path, cert.state.report_txt_path
+ cert.state.report.pdf_path, cert.state.report.txt_path
)
# If OCR was done the result was garbage
- cert.state.report_convert_garbage = ocr_done
+ cert.state.report.convert_garbage = ocr_done
# And put the whole result into convert_ok
- cert.state.report_convert_ok = ok_result
+ cert.state.report.convert_ok = ok_result
if not ok_result:
error_msg = "failed to convert report pdf->txt"
logger.error(f"Cert dgst: {cert.dgst} " + error_msg)
else:
- cert.state.report_txt_hash = helpers.get_sha256_filepath(cert.state.report_txt_path)
+ cert.state.report.txt_hash = helpers.get_sha256_filepath(cert.state.report.txt_path)
return cert
@staticmethod
@@ -879,34 +835,39 @@ class CCCertificate(
"""
Converts the pdf security target to txt, given the certificate. Staticmethod to allow for parallelization.
- :param CCCertificate cert: cert to download the pdf security target for
+ :param CCCertificate cert: cert to convert the pdf security target for
:return CCCertificate: the modified certificate with updated state
"""
- ocr_done, ok_result = sec_certs.utils.pdf.convert_pdf_file(cert.state.st_pdf_path, cert.state.st_txt_path)
+ ocr_done, ok_result = sec_certs.utils.pdf.convert_pdf_file(cert.state.st.pdf_path, cert.state.st.txt_path)
# If OCR was done the result was garbage
- cert.state.st_convert_garbage = ocr_done
+ cert.state.st.convert_garbage = ocr_done
# And put the whole result into convert_ok
- cert.state.st_convert_ok = ok_result
+ cert.state.st.convert_ok = ok_result
if not ok_result:
error_msg = "failed to convert security target pdf->txt"
logger.error(f"Cert dgst: {cert.dgst} " + error_msg)
else:
- cert.state.st_txt_hash = helpers.get_sha256_filepath(cert.state.st_txt_path)
+ cert.state.st.txt_hash = helpers.get_sha256_filepath(cert.state.st.txt_path)
return cert
@staticmethod
- def extract_st_pdf_metadata(cert: CCCertificate) -> CCCertificate:
+ def convert_cert_pdf(cert: CCCertificate) -> CCCertificate:
"""
- Extracts metadata from security target pdf given the certificate. Staticmethod to allow for parallelization.
+ Converts the pdf certificate to txt, given the certificate. Staticmethod to allow for parallelization.
- :param CCCertificate cert: cert to extract the metadata for.
+ :param CCCertificate cert: cert to convert the certificate for
:return CCCertificate: the modified certificate with updated state
"""
- response, cert.pdf_data.st_metadata = sec_certs.utils.pdf.extract_pdf_metadata(cert.state.st_pdf_path)
- if response != constants.RETURNCODE_OK:
- cert.state.st_extract_ok = False
+ ocr_done, ok_result = sec_certs.utils.pdf.convert_pdf_file(cert.state.cert.pdf_path, cert.state.cert.txt_path)
+ # If OCR was done the result was garbage
+ cert.state.cert.convert_garbage = ocr_done
+ # And put the whole result into convert_ok
+ cert.state.cert.convert_ok = ok_result
+ if not ok_result:
+ error_msg = "failed to convert security target pdf->txt"
+ logger.error(f"Cert dgst: {cert.dgst} " + error_msg)
else:
- cert.state.st_extract_ok = True
+ cert.state.cert.txt_hash = helpers.get_sha256_filepath(cert.state.cert.txt_path)
return cert
@staticmethod
@@ -917,28 +878,41 @@ class CCCertificate(
:param CCCertificate cert: cert to extract the metadata for.
:return CCCertificate: the modified certificate with updated state
"""
- response, cert.pdf_data.report_metadata = sec_certs.utils.pdf.extract_pdf_metadata(cert.state.report_pdf_path)
+ response, cert.pdf_data.report_metadata = sec_certs.utils.pdf.extract_pdf_metadata(cert.state.report.pdf_path)
if response != constants.RETURNCODE_OK:
- cert.state.report_extract_ok = False
+ cert.state.report.extract_ok = False
else:
- cert.state.report_extract_ok = True
+ cert.state.report.extract_ok = True
return cert
@staticmethod
- def extract_st_pdf_frontpage(cert: CCCertificate) -> CCCertificate:
+ def extract_st_pdf_metadata(cert: CCCertificate) -> CCCertificate:
"""
- Extracts data from security target pdf frontpage given the certificate. Staticmethod to allow for parallelization.
+ Extracts metadata from security target pdf given the certificate. Staticmethod to allow for parallelization.
- :param CCCertificate cert: cert to extract the frontpage data for.
+ :param CCCertificate cert: cert to extract the metadata for.
:return CCCertificate: the modified certificate with updated state
"""
- cert.pdf_data.st_frontpage = {}
+ response, cert.pdf_data.st_metadata = sec_certs.utils.pdf.extract_pdf_metadata(cert.state.st.pdf_path)
+ if response != constants.RETURNCODE_OK:
+ cert.state.st.extract_ok = False
+ else:
+ cert.state.st.extract_ok = True
+ return cert
- for header_type, associated_header_func in HEADERS.items():
- response, cert.pdf_data.st_frontpage[header_type] = associated_header_func(cert.state.st_txt_path)
+ @staticmethod
+ def extract_cert_pdf_metadata(cert: CCCertificate) -> CCCertificate:
+ """
+ Extracts metadata from certificate pdf given the certificate. Staticmethod to allow for parallelization.
- if response != constants.RETURNCODE_OK:
- cert.state.st_extract_ok = False
+ :param CCCertificate cert: cert to extract the metadata for.
+ :return CCCertificate: the modified certificate with updated state
+ """
+ response, cert.pdf_data.cert_metadata = sec_certs.utils.pdf.extract_pdf_metadata(cert.state.cert.pdf_path)
+ if response != constants.RETURNCODE_OK:
+ cert.state.cert.extract_ok = False
+ else:
+ cert.state.cert.extract_ok = True
return cert
@staticmethod
@@ -951,25 +925,25 @@ class CCCertificate(
"""
cert.pdf_data.report_frontpage = {}
- for header_type, associated_header_func in HEADERS.items():
- response, cert.pdf_data.report_frontpage[header_type] = associated_header_func(cert.state.report_txt_path)
-
+ if cert.scheme in scheme_frontpage_functions:
+ header_func = scheme_frontpage_functions[cert.scheme]
+ response, cert.pdf_data.report_frontpage[cert.scheme] = header_func(cert.state.report.txt_path)
if response != constants.RETURNCODE_OK:
- cert.state.report_extract_ok = False
+ cert.state.report.extract_ok = False
return cert
@staticmethod
def extract_report_pdf_keywords(cert: CCCertificate) -> CCCertificate:
"""
- Matches regular expresions in txt obtained from certification report and extracts the matches into attribute.
+ Matches regular expressions in txt obtained from certification report and extracts the matches into attribute.
Static method to allow for parallelization
:param CCCertificate cert: certificate to extract the keywords for.
:return CCCertificate: the modified certificate with extracted keywords.
"""
- report_keywords = sec_certs.utils.extract.extract_keywords(cert.state.report_txt_path, cc_rules)
+ report_keywords = sec_certs.utils.extract.extract_keywords(cert.state.report.txt_path, cc_rules)
if report_keywords is None:
- cert.state.report_extract_ok = False
+ cert.state.report.extract_ok = False
else:
cert.pdf_data.report_keywords = report_keywords
return cert
@@ -977,19 +951,35 @@ class CCCertificate(
@staticmethod
def extract_st_pdf_keywords(cert: CCCertificate) -> CCCertificate:
"""
- Matches regular expresions in txt obtained from security target and extracts the matches into attribute.
+ Matches regular expressions in txt obtained from security target and extracts the matches into attribute.
Static method to allow for parallelization
:param CCCertificate cert: certificate to extract the keywords for.
:return CCCertificate: the modified certificate with extracted keywords.
"""
- st_keywords = sec_certs.utils.extract.extract_keywords(cert.state.st_txt_path, cc_rules)
+ st_keywords = sec_certs.utils.extract.extract_keywords(cert.state.st.txt_path, cc_rules)
if st_keywords is None:
- cert.state.st_extract_ok = False
+ cert.state.st.extract_ok = False
else:
cert.pdf_data.st_keywords = st_keywords
return cert
+ @staticmethod
+ def extract_cert_pdf_keywords(cert: CCCertificate) -> CCCertificate:
+ """
+ Matches regular expressions in txt obtained from the certificate and extracts the matches into attribute.
+ Static method to allow for parallelization
+
+ :param CCCertificate cert: certificate to extract the keywords for.
+ :return CCCertificate: the modified certificate with extracted keywords.
+ """
+ cert_keywords = sec_certs.utils.extract.extract_keywords(cert.state.cert.txt_path, cc_rules)
+ if cert_keywords is None:
+ cert.state.cert.extract_ok = False
+ else:
+ cert.pdf_data.cert_keywords = cert_keywords
+ return cert
+
def compute_heuristics_version(self) -> None:
"""
Fills in the heuristically obtained version of certified product into attribute in heuristics class.
diff --git a/src/sec_certs/sample/cc_certificate_id.py b/src/sec_certs/sample/cc_certificate_id.py
index 254c8e05..c2b6daa2 100644
--- a/src/sec_certs/sample/cc_certificate_id.py
+++ b/src/sec_certs/sample/cc_certificate_id.py
@@ -2,133 +2,224 @@ from __future__ import annotations
import re
from dataclasses import dataclass
+from functools import cached_property
+from sec_certs.cert_rules import rules
-@dataclass(eq=True, frozen=True)
-class CertificateId:
- """
- A Common Criteria certificate id.
- """
- scheme: str
- raw: str
+def _parse_year(year: str | None) -> int | None:
+ if year is None:
+ return None
+ y = int(year)
+ if y < 50:
+ return y + 2000
+ elif y < 100:
+ return y + 1900
+ else:
+ return y
+
+
+def FR(meta) -> str:
+ year = _parse_year(meta["year"])
+ counter = meta["counter"]
+ doc = meta.get("doc")
+ version = meta.get("version")
+ cert_id = f"ANSSI-CC-{year}/{counter}"
+ if doc:
+ cert_id += f"-{doc}"
+ if version:
+ cert_id += f"v{version}"
+ return cert_id
+
+
+def DE(meta) -> str:
+ s = meta.get("s")
+ counter = meta["counter"]
+ version = meta.get("version")
+ year = _parse_year(meta.get("year"))
+ doc = meta.get("doc")
+ cert_id = "BSI-DSZ-CC"
+ if s:
+ cert_id += f"-{s}"
+ cert_id += f"-{counter}"
+ if version:
+ cert_id += f"-{version.upper()}"
+ if year:
+ cert_id += f"-{year}"
+ if doc:
+ cert_id += f"-{doc}"
+ return cert_id
+
+
+def US(meta) -> str:
+ counter = meta["counter"]
+ cc = meta.get("cc")
+ vid = meta.get("VID")
+ year = _parse_year(meta.get("year"))
+ cert_id = "CCEVS-VR"
+ if cc:
+ cert_id += f"-{cc}"
+ if vid:
+ cert_id += f"-{vid}"
+ cert_id += f"-{counter}"
+ if year:
+ cert_id += f"-{year}"
+ return cert_id
+
+
+def MY(meta) -> str:
+ digit = meta["digit"]
+ counter = meta["counter"]
+ version = meta["version"]
+ return f"ISCB-{digit}-RPT-C{counter}-CR-{version.lower()}"
+
+
+def ES(meta) -> str:
+ year = _parse_year(meta["year"])
+ project = meta["project"]
+ counter = meta["counter"]
+ # Version is intentionally cut here, as it seems to refer to an internal version of the report.
+ # version = groups["version"]
+ return f"{year}-{project}-INF-{counter}"
- def _canonical_fr(self) -> str:
- def pad_last_segment_with_zero(id_str: str) -> str:
- splitted = id_str.split("/")
- if len(splitted) > 1:
- num = splitted[-1].zfill(2)
- return f"{''.join(splitted[:-1])}/{num}"
- return id_str
- new_cert_id = self.clean
- rules = [
- "(?:Rapport de certification|Certification Report) ([0-9]+[/-_][0-9]+(?:[vV][1-9])?(?:[_/-][MSR][0-9]+)?)",
- "(?:ANSS[Ii]|DCSSI)(?:-CC)?[- ]([0-9]+[/-_][0-9]+(?:[vV][1-9])?(?:[_/-][MSR][0-9]+)?)",
- "([0-9]+[/-_][0-9]+(?:[vV][1-9])?(?:[_/-][MSR][0-9]+)?)",
- ]
- for rule in rules:
- if match := re.match(rule, new_cert_id):
- return pad_last_segment_with_zero("ANSSI-CC-" + match.group(1).replace("_", "/").replace("V", "v"))
+def IN(meta) -> str:
+ lab = meta["lab"]
+ vendor = meta["vendor"]
+ level = meta["level"]
+ number1, number2 = meta["number1"], meta["number2"]
+ return f"IC3S/{lab}/{vendor}/{level}/{number1}/{number2}"
- return new_cert_id
- def _canonical_de(self) -> str:
- def extract_parts(bsi_parts: list[str]) -> tuple:
- cert_num = None
- cert_version = None
- cert_year = None
+def SE(meta) -> str:
+ year = _parse_year(meta["year"])
+ counter = int(meta["counter"])
+ return f"CSEC{year}{counter:03}"
- if len(bsi_parts) > 3:
- cert_num = bsi_parts[3]
- if len(bsi_parts) > 4:
- if bsi_parts[4].startswith("V") or bsi_parts[4].startswith("v"):
- cert_version = bsi_parts[4].upper() # get version in uppercase
- else:
- cert_year = bsi_parts[4]
- if len(bsi_parts) > 5:
- cert_year = bsi_parts[5]
- return cert_num, cert_version, cert_year
+def UK(meta) -> str:
+ counter = meta["counter"]
+ return f"CRP{counter}"
- bsi_parts = self.clean.split("-")
- cert_num, cert_version, cert_year = extract_parts(bsi_parts)
+def CA(meta) -> str:
+ if "lab" in meta:
+ year = _parse_year(meta.get("year"))
+ number = meta["number"]
+ lab = meta["lab"]
+ cert_id = f"{number}-{lab}"
+ if year:
+ cert_id += f"-{year}"
+ return cert_id
+ else:
+ number1 = meta["number1"]
+ digit = meta["digit"]
+ number2 = meta["number2"]
+ return f"{number1}-{digit}-{number2}"
- # reconstruct BSI number again
- new_cert_id = "BSI-DSZ-CC"
- if cert_num is not None:
- new_cert_id += "-" + cert_num
- if cert_version is not None:
- new_cert_id += "-" + cert_version
- if cert_year is not None:
- new_cert_id += "-" + cert_year
- return new_cert_id
+def JP(meta) -> str:
+ counter = meta["counter"]
+ digit = meta.get("digit")
+ year = _parse_year(meta.get("year"))
+ cert_id = f"JISEC-CC-CRP-C{counter}"
+ if digit:
+ cert_id += f"-{digit}"
+ if year:
+ cert_id += f"-{year}"
+ return cert_id
- def _canonical_es(self) -> str:
- cert_id = self.clean
- spain_parts = cert_id.split("-")
- cert_year = spain_parts[0]
- cert_batch = spain_parts[1].lstrip("0")
- cert_num = spain_parts[3].lstrip("0")
- if "v" in cert_num:
- cert_num = cert_num[: cert_num.find("v")]
- if "V" in cert_num:
- cert_num = cert_num[: cert_num.find("V")]
+def KR(meta) -> str:
+ word = meta["word"]
+ counter = int(meta["counter"])
+ year = _parse_year(meta["year"])
+ return f"KECS-{word}-{counter:04}-{year}"
- new_cert_id = f"{cert_year}-{cert_batch}-INF-{cert_num.strip()}" # drop version # TODO: Maybe do not drop?
- return new_cert_id
+def TR(meta) -> str:
+ prefix = meta["prefix"]
+ number = meta["number"]
+ return f"{prefix}/TSE-CCCS-{number}"
- def _canonical_it(self):
- new_cert_id = self.clean
- if not new_cert_id.endswith("/RC"):
- new_cert_id = new_cert_id + "/RC"
- return new_cert_id
+def NO(meta) -> str:
+ counter = int(meta["counter"])
+ return f"SERTIT-{counter:03}"
- def _canonical_in(self):
- return self.clean.replace(" ", "")
- def _canonical_se(self):
- return self.clean.replace(" ", "")
+def NL(meta) -> str:
+ core = meta["core"]
+ doc = meta.get("doc")
+ if doc is None:
+ doc = "CR"
+ return f"NSCIB-CC-{core}-{doc}"
- def _canonical_uk(self):
- new_cert_id = self.clean
- if match := re.match("CERTIFICATION REPORT No. P([0-9]+[A-Z]?)", new_cert_id):
- new_cert_id = "CRP" + match.group(1)
- return new_cert_id
- def _canonical_ca(self):
- new_cert_id = self.clean
- if new_cert_id.endswith("-CR"):
- new_cert_id = new_cert_id[:-3]
- if new_cert_id.endswith("P"):
- new_cert_id = new_cert_id[:-1]
- return new_cert_id.replace(" ", "-")
+def AU(meta) -> str:
+ counter = meta["counter"]
+ year_s = meta["year"]
+ if len(year_s) < len(counter):
+ # Hack for some mistakes in their ordering
+ year_s, counter = counter, year_s
+ year = _parse_year(year_s)
+ return f"Certificate Number: {year}/{counter}"
- def _canonical_jp(self):
- new_cert_id = self.clean
- if match := re.match("Certification No. (C[0-9]+)", new_cert_id):
- return match.group(1)
- if match := re.search("CRP-(C[0-9]+)-", new_cert_id):
- return match.group(1)
- return new_cert_id
- def _canonical_no(self):
- new_cert_id = self.clean
- cert_num = int(new_cert_id.split("-")[1])
- return f"SERTIT-{cert_num:03}"
+def SG(meta) -> str:
+ year = meta["year"]
+ counter = meta["counter"]
+ return f"CSA_CC_{year}{counter}"
- def _canonical_nl(self):
- new_cert_id = self.clean
- if new_cert_id.startswith("CC-"):
- new_cert_id = f"NSCIB-{new_cert_id}"
- if not new_cert_id.endswith("-CR"):
- new_cert_id = f"{new_cert_id}-CR"
- return new_cert_id
+
+def IT(meta) -> str:
+ lab = meta.get("lab")
+ counter = meta["counter"]
+ year = _parse_year(meta["year"])
+ cert_id = "OCSI/CERT/"
+ if lab:
+ cert_id += f"{lab}/"
+ cert_id += f"{counter}/{year}/RC"
+ return cert_id
+
+
+# We have rules for some schemes to make canonical cert_ids.
+schemes = {
+ "FR": FR,
+ "DE": DE,
+ "US": US,
+ "MY": MY,
+ "ES": ES,
+ "IN": IN,
+ "SE": SE,
+ "UK": UK,
+ "CA": CA,
+ "JP": JP,
+ "NO": NO,
+ "NL": NL,
+ "AU": AU,
+ "KR": KR,
+ "TR": TR,
+ "SG": SG,
+ "IT": IT,
+}
+
+
+@dataclass(frozen=True)
+class CertificateId:
+ """
+ A Common Criteria certificate id.
+ """
+
+ scheme: str
+ raw: str
+
+ @cached_property
+ def meta(self):
+ for rule in rules["cc_cert_id"][self.scheme]:
+ if match := re.match(rule, self.clean):
+ return match.groupdict()
+ return {}
@property
def clean(self) -> str:
@@ -142,25 +233,25 @@ class CertificateId:
"""
The canonical version of this certificate id.
"""
- # We have rules for some schemes to make canonical cert_ids.
- schemes = {
- "FR": self._canonical_fr,
- "DE": self._canonical_de,
- "ES": self._canonical_es,
- "IT": self._canonical_it,
- "IN": self._canonical_in,
- "SE": self._canonical_se,
- "UK": self._canonical_uk,
- "CA": self._canonical_ca,
- "JP": self._canonical_jp,
- "NO": self._canonical_no,
- "NL": self._canonical_nl,
- }
+ clean = self.clean
if self.scheme in schemes:
- return schemes[self.scheme]()
+ return schemes[self.scheme](self.meta)
else:
- return self.clean
+ return clean
+
+ def __str__(self):
+ return self.canonical
+
+ def __hash__(self):
+ return hash((self.scheme, self.raw))
+
+ def __eq__(self, other):
+ if isinstance(other, str):
+ return self.canonical == other
+ if not isinstance(other, CertificateId):
+ return False
+ return self.canonical == other.canonical and self.scheme == other.scheme
def canonicalize(cert_id_str: str, scheme: str) -> str:
diff --git a/src/sec_certs/sample/cc_scheme.py b/src/sec_certs/sample/cc_scheme.py
index 73946cf3..8d18b5d0 100644
--- a/src/sec_certs/sample/cc_scheme.py
+++ b/src/sec_certs/sample/cc_scheme.py
@@ -4,6 +4,7 @@ from __future__ import annotations
import hashlib
import math
+import re
import tempfile
import warnings
from collections.abc import Callable, Iterable
@@ -30,6 +31,7 @@ __all__ = [
"get_canada_certified",
"get_canada_in_evaluation",
"get_france_certified",
+ "get_france_archived",
"get_germany_certified",
"get_india_certified",
"get_india_archived",
@@ -67,7 +69,13 @@ def _get(url: str, session, **kwargs) -> Response:
with warnings.catch_warnings():
warnings.simplefilter("ignore", category=InsecureRequestWarning)
conn = session if session else requests
- resp = conn.get(url, headers={"User-Agent": "seccerts.org"}, verify=False, **kwargs)
+ resp = conn.get(
+ url,
+ headers={"User-Agent": "seccerts.org"},
+ verify=False,
+ **kwargs,
+ timeout=10,
+ )
resp.raise_for_status()
return resp
@@ -84,13 +92,16 @@ def _get_hash(url: str, session=None) -> bytes:
return h.digest()
-def get_australia_in_evaluation(enhanced: bool = True) -> list[dict[str, Any]]: # noqa: C901
+def get_australia_in_evaluation( # noqa: C901
+ enhanced: bool = True,
+) -> list[dict[str, Any]]:
"""
Get Australia "products in evaluation" entries.
:param enhanced: Whether to enhance the results by following links (slower, more data).
:return: The entries.
"""
+ # TODO: Australian scheme is blocking our User-Agent.
soup = _get_page(constants.CC_AUSTRALIA_INEVAL_URL)
header = soup.find("h2", string="Products in evaluation")
table = header.find_next_sibling("table")
@@ -154,7 +165,10 @@ def get_canada_certified() -> list[dict[str, Any]]:
:return: The entries.
"""
- resp = _get(constants.CC_CANADA_API_URL + f"?lang=en&url={constants.CC_CANADA_CERTIFIED_URL}", None)
+ resp = _get(
+ constants.CC_CANADA_API_URL + f"?lang=en&url={constants.CC_CANADA_CERTIFIED_URL}",
+ None,
+ )
html_data = resp.json()["response"]["page"]["body"][0]
soup = BeautifulSoup(html_data, "html5lib")
tbody = soup.find("table").find("tbody")
@@ -179,7 +193,10 @@ def get_canada_in_evaluation() -> list[dict[str, Any]]:
:return: The entries.
"""
- resp = _get(constants.CC_CANADA_API_URL + f"?lang=en&url={constants.CC_CANADA_INEVAL_URL}", None)
+ resp = _get(
+ constants.CC_CANADA_API_URL + f"?lang=en&url={constants.CC_CANADA_INEVAL_URL}",
+ None,
+ )
html_data = resp.json()["response"]["page"]["body"][0]
soup = BeautifulSoup(html_data, "html5lib")
tbody = soup.find("table").find("tbody")
@@ -198,100 +215,114 @@ def get_canada_in_evaluation() -> list[dict[str, Any]]:
return results
-def get_france_certified(enhanced: bool = True, artifacts: bool = False) -> list[dict[str, Any]]: # noqa: C901
- """
- Get French "certified product" 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.
- """
- base_soup = _get_page(constants.CC_ANSSI_CERTIFIED_URL)
- category_nav = base_soup.find("ul", class_="nav-categories")
+def _get_france(url, enhanced, artifacts) -> list[dict[str, Any]]: # noqa: C901
+ base_soup = _get_page(url)
+ pager = base_soup.find("nav", class_="pager")
+ last_page_a = re.search("[0-9]+", pager.find("a", title="Aller à la dernière page").text)
+ if not last_page_a:
+ raise ValueError
+ pages = int(last_page_a.group())
results = []
- for li in tqdm(category_nav.find_all("li"), desc="Get FR scheme certified."):
- a = li.find("a")
- url = a["href"]
- category_name = sns(a.text)
- soup = _get_page(urljoin(constants.CC_ANSSI_BASE_URL, url))
- table = soup.find("table", class_="produits-liste cc")
- if not table:
- continue
- tbody = table.find("tbody")
- for tr in tqdm(tbody.find_all("tr")):
- tds = tr.find_all("td")
- if not tds:
- continue
+ for page in range(pages + 1):
+ soup = _get_page(url + f"?page={page}")
+ for row in soup.find_all("article", class_="node--type-produit-certifie-cc"):
cert: dict[str, Any] = {
- "product": sns(tds[0].text),
- "vendor": sns(tds[1].text),
- "level": sns(tds[2].text),
- "id": sns(tds[3].text),
- "certification_date": sns(tds[4].text),
- "category": category_name,
- "url": urljoin(constants.CC_ANSSI_BASE_URL, tds[0].find("a")["href"]),
+ "product": sns(row.find("h3").text),
+ "url": urljoin(constants.CC_ANSSI_BASE_URL, row.find("a")["href"]),
}
+ description_para = row.find("p", class_="field-body")
+ if description_para:
+ cert["description"] = sns(description_para.text)
+ complement_info = row.find("div", class_="info-complement")
+ for li in complement_info.find_all("li"):
+ label = li.find("span").text
+ value = sns(li.find(string=True, recursive=False))
+ if "Commanditaire" in label:
+ cert["sponsor"] = value
+ elif "Développeur" in label:
+ cert["developer"] = value
+ elif "Référence du certificat" in label:
+ cert["cert_id"] = value
+ elif "Niveau" in label:
+ cert["level"] = value
+ elif "Date de fin de validité" in label:
+ cert["expiration_date"] = value
if enhanced:
e: dict[str, Any] = {}
cert_page = _get_page(cert["url"])
- ref = cert_page.find("div", class_="ref-date")
- for ref_li in ref.find_all("li"):
- title, value = (sns(span.text) for span in ref_li.find_all("span", recursive=False))
- if not title:
- continue
- if "Référence" in title:
- e["id"] = value
- elif "Date de certification" in title:
+ infos = cert_page.find("div", class_="product-infos-wrapper")
+ for tr in infos.find_all("tr"):
+ label = tr.find("th").text
+ value = sns(tr.find("td").text)
+ if "Référence du certificat" in label:
+ e["cert_id"] = value
+ elif "Date de certification" in label:
e["certification_date"] = value
- elif "Date de fin de validité" in title:
+ elif "Date de fin de validité" in label:
e["expiration_date"] = value
- details = cert_page.find("div", class_="details")
- for detail_li in details.find_all("li"):
- title, value = (sns(span.text) for span in detail_li.find_all("span", recursive=False))
- if not title:
- continue
- if "Catégorie" in title:
+ elif "Catégorie" in label:
+ # TODO: translate?
e["category"] = value
- elif "Référentiel" in title:
+ elif "Référentiel" in label:
e["cc_version"] = value
- elif "Niveau" in title:
- e["level"] = value
- elif "Augmentations" in title:
- e["augmentations"] = value
- elif "Profil de protection" in title:
- e["protection_profile"] = value
- elif "Développeur" in title:
+ elif "Développeur(s)" in label:
e["developer"] = value
- elif "Centre d'évaluation" in title:
+ elif "Commanditaire(s)" in label:
+ e["sponsor"] = value
+ elif "Centre d'évaluation" in label:
e["evaluation_facility"] = value
- elif "Accords de reconnaissance" in title:
- e["recognition"] = value
- e["description"] = sns(cert_page.find("div", class_="box-produit-descriptif").text)
- links = cert_page.find("div", class_="box-produit-telechargements")
- for link_li in links.find_all("li"):
- a = link_li.find("a")
- href = urljoin(constants.CC_ANSSI_BASE_URL, a["href"])
- title = sns(a.text)
- if not title:
- continue
- if "Rapport de certification" in title:
- e["report_link"] = href
+ elif "Niveau" in label:
+ e["level"] = value
+ elif "Profil de protection" in label:
+ e["protection_profile"] = value
+ elif "Accords de reconnaissance" in label:
+ e["mutual_recognition"] = value
+ elif "Augmentations" in label:
+ e["augmented"] = value
+ documents = cert_page.find("div", class_="documents")
+ for a in documents.find_all("a"):
+ if "Rapport de certification" in a.text:
+ e["report_link"] = urljoin(constants.CC_ANSSI_BASE_URL, a["href"])
if artifacts:
- e["report_hash"] = _get_hash(href).hex()
- elif "Security target" in title:
- e["target_link"] = href
+ e["report_hash"] = _get_hash(e["report_link"]).hex()
+ elif "Cible de sécurité" in a.text:
+ e["target_link"] = urljoin(constants.CC_ANSSI_BASE_URL, a["href"])
if artifacts:
- e["target_hash"] = _get_hash(href).hex()
- elif "Certificat" in title:
- e["cert_link"] = href
+ e["target_hash"] = _get_hash(e["target_link"]).hex()
+ elif "Certificat" in a.text:
+ e["cert_link"] = urljoin(constants.CC_ANSSI_BASE_URL, a["href"])
if artifacts:
- e["cert_hash"] = _get_hash(href).hex()
+ e["cert_hash"] = _get_hash(e["cert_link"]).hex()
cert["enhanced"] = e
results.append(cert)
return results
-def get_germany_certified(enhanced: bool = True, artifacts: bool = False) -> list[dict[str, Any]]: # noqa: C901
+def get_france_certified(enhanced: bool = True, artifacts: bool = False) -> list[dict[str, Any]]: # noqa: C901
+ """
+ Get French "certified product" 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.
+ """
+ return _get_france(constants.CC_ANSSI_CERTIFIED_URL, enhanced, artifacts)
+
+
+def get_france_archived(enhanced: bool = True, artifacts: bool = False) -> list[dict[str, Any]]: # noqa: C901
+ """
+ Get French "archived product" 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.
+ """
+ return _get_france(constants.CC_ANSSI_ARCHIVED_URL, enhanced, artifacts)
+
+
+def get_germany_certified( # noqa: C901
+ enhanced: bool = True, artifacts: bool = False
+) -> list[dict[str, Any]]:
"""
Get German "certified product" entries.
@@ -731,40 +762,99 @@ def get_japan_in_evaluation() -> list[dict[str, Any]]:
return results
-def get_malaysia_certified() -> list[dict[str, Any]]:
- """
- Get Malaysian "certified product" entries.
-
- :return: The entries.
- """
- soup = _get_page(constants.CC_MALAYSIA_CERTIFIED_URL)
- sections = soup.find("div", attrs={"itemprop": "articleBody"}).find_all("section", class_="sppb-section")
+def _get_malaysia(url, enhanced, artifacts) -> list[dict[str, Any]]: # noqa: C901
+ soup = _get_page(url)
+ pages_re = re.search("Page [0-9]+ of ([0-9]+)", soup.find("form").text)
+ if not pages_re:
+ raise ValueError
+ total_pages = int(pages_re.group(1))
results = []
- for section in sections:
- table = section.find("table")
- if table is None:
- continue
- heading = section.find("h5")
- if heading is None:
- continue
- category_name = sns(heading.text)
- tbody = table.find("tbody")
- for tr in tbody.find_all("tr", recursive=False):
- tds = tr.find_all("td", recursive=False)
- if len(tds) != 6:
- continue
- cert = {
- "category": category_name,
- "level": sns(tds[0].text),
- "cert_id": sns(tds[1].text),
- "certification_date": sns(tds[2].text),
+ for i in range(total_pages):
+ soup = _get_page(url + f"?start={i * 10}")
+ table = soup.find("table", class_="directoryTable")
+ for tr in table.find_all("tr", class_="directoryRow"):
+ tds = tr.find_all("td")
+ cert: dict[str, Any] = {
+ "cert_no": sns(tds[0].text),
+ "developer": sns(tds[1].text),
+ "level": sns(tds[2].text),
"product": sns(tds[3].text),
- "developer": sns(tds[4].text),
+ "certification_date": sns(tds[4].text),
+ "expiration_date": sns(tds[5].text),
+ "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"])
+ 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")
+ title = left.text
+ value = sns(right.text)
+ if "Project ID" in title:
+ e["cert_id"] = value
+ elif "Product Name and Version" in title:
+ e["product"] = sns(right.text)
+ elif "Product Sponsor / Developer" in title:
+ e["developer"] = value
+ elif "Category" in title:
+ e["category"] = value
+ elif "Product Type" in title:
+ e["type"] = value
+ elif "Scope" in title:
+ e["scope"] = value
+ elif "Product Sponsor / Developer Contact Details" in title:
+ e["developer_contact"] = value
+ elif "Assurance Level" in title:
+ e["assurance_level"] = value
+ elif "Certificate Date" in title:
+ e["certification_date"] = value
+ elif "Expiry Date" in title:
+ e["expiration_date"] = value
+ elif "Recognized By" in title:
+ e["mutual_recognition"] = value
+ elif "Reports" in title:
+ for a in right.find_all("a"):
+ 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"]).hex()
+ 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"]).hex()
+ elif "Maintenance" in title:
+ pass
+ elif "Status" in title:
+ e["status"] = value
+ cert["enhanced"] = e
results.append(cert)
return results
+def get_malaysia_certified(enhanced: bool = True, artifacts: bool = False) -> list[dict[str, Any]]:
+ """
+ Get Malaysian "certified product" 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.
+ """
+ return _get_malaysia(constants.CC_MALAYSIA_CERTIFIED_URL, enhanced, artifacts)
+
+
+def get_malaysia_archived(enhanced: bool = True, artifacts: bool = False) -> list[dict[str, Any]]:
+ """
+ Get Malaysian "archived product" 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.
+ """
+ return _get_malaysia(constants.CC_MALAYSIA_ARCHIVED_URL, enhanced, artifacts)
+
+
def get_malaysia_in_evaluation() -> list[dict[str, Any]]:
"""
Get Malaysian "product in evaluation" entries.
@@ -790,7 +880,9 @@ def get_malaysia_in_evaluation() -> list[dict[str, Any]]:
return results
-def get_netherlands_certified(artifacts: bool = False) -> list[dict[str, Any]]: # noqa: C901
+def get_netherlands_certified( # noqa: C901
+ artifacts: bool = False,
+) -> list[dict[str, Any]]:
"""
Get Dutch "certified product" entries.
@@ -860,7 +952,9 @@ def get_netherlands_in_evaluation() -> list[dict[str, Any]]:
return results
-def _get_norway(url: str, enhanced: bool, artifacts: bool) -> list[dict[str, Any]]: # noqa: C901
+def _get_norway( # noqa: C901
+ url: str, enhanced: bool, artifacts: bool
+) -> list[dict[str, Any]]:
soup = _get_page(url)
results = []
for tr in soup.find_all("tr", class_="certified-product"):
@@ -957,7 +1051,9 @@ def get_norway_archived(enhanced: bool = True, artifacts: bool = False) -> list[
return _get_norway(constants.CC_NORWAY_ARCHIVED_URL, enhanced, artifacts)
-def _get_korea(product_class: int, enhanced: bool, artifacts: bool) -> list[dict[str, Any]]: # noqa: C901
+def _get_korea( # noqa: C901
+ product_class: int, enhanced: bool, artifacts: bool
+) -> list[dict[str, Any]]:
session = requests.session()
session.get(constants.CC_KOREA_EN_URL)
# Get base page
@@ -1121,7 +1217,10 @@ def _get_singapore(url: str, artifacts: bool) -> list[dict[str, Any]]:
"cert_title": obj["certificate"]["title"],
"cert_link": urljoin(constants.CC_SINGAPORE_BASE_URL, obj["certificate"]["mediaUrl"]),
"report_title": obj["certificationReport"]["title"],
- "report_link": urljoin(constants.CC_SINGAPORE_BASE_URL, obj["certificationReport"]["mediaUrl"]),
+ "report_link": urljoin(
+ constants.CC_SINGAPORE_BASE_URL,
+ obj["certificationReport"]["mediaUrl"],
+ ),
"target_title": obj["securityTarget"]["title"],
"target_link": urljoin(constants.CC_SINGAPORE_BASE_URL, obj["securityTarget"]["mediaUrl"]),
}
@@ -1213,7 +1312,9 @@ def get_spain_certified() -> list[dict[str, Any]]:
return results
-def _get_sweden(url: str, enhanced: bool, artifacts: bool) -> list[dict[str, Any]]: # noqa: C901
+def _get_sweden( # noqa: C901
+ url: str, enhanced: bool, artifacts: bool
+) -> list[dict[str, Any]]:
soup = _get_page(url)
nav = soup.find("main").find("nav", class_="component-nav-box__list")
results = []
@@ -1319,8 +1420,7 @@ def get_turkey_certified() -> list[dict[str, Any]]:
with tempfile.TemporaryDirectory() as tmpdir:
pdf_path = Path(tmpdir) / "turkey.pdf"
resp = requests.get(constants.CC_TURKEY_ARCHIVED_URL)
- if resp.status_code != requests.codes.ok:
- raise ValueError(f"Unable to download: status={resp.status_code}")
+ resp.raise_for_status()
with pdf_path.open("wb") as f:
f.write(resp.content)
dfs = tabula.read_pdf(str(pdf_path), pages="all")
@@ -1344,7 +1444,9 @@ def get_turkey_certified() -> list[dict[str, Any]]:
return results
-def get_usa_certified(enhanced: bool = True, artifacts: bool = False) -> list[dict[str, Any]]: # noqa: C901
+def get_usa_certified( # noqa: C901
+ enhanced: bool = True, artifacts: bool = False
+) -> list[dict[str, Any]]:
"""
Get American "certified product" entries.
@@ -1521,20 +1623,45 @@ class CCScheme(ComplexSerializableType):
methods: ClassVar[dict[str, dict[EntryType, Callable]]] = {
"AU": {EntryType.InEvaluation: get_australia_in_evaluation},
- "CA": {EntryType.InEvaluation: get_canada_in_evaluation, EntryType.Certified: get_canada_certified},
- "FR": {EntryType.Certified: get_france_certified},
+ "CA": {
+ EntryType.InEvaluation: get_canada_in_evaluation,
+ EntryType.Certified: get_canada_certified,
+ },
+ "FR": {
+ EntryType.Certified: get_france_certified,
+ EntryType.Archived: get_france_archived,
+ },
"DE": {EntryType.Certified: get_germany_certified},
- "IN": {EntryType.Certified: get_india_certified, EntryType.Archived: get_india_archived},
- "IT": {EntryType.Certified: get_italy_certified, EntryType.InEvaluation: get_italy_in_evaluation},
+ "IN": {
+ EntryType.Certified: get_india_certified,
+ EntryType.Archived: get_india_archived,
+ },
+ "IT": {
+ EntryType.Certified: get_italy_certified,
+ EntryType.InEvaluation: get_italy_in_evaluation,
+ },
"JP": {
EntryType.InEvaluation: get_japan_in_evaluation,
EntryType.Certified: get_japan_certified,
EntryType.Archived: get_japan_archived,
},
- "MY": {EntryType.Certified: get_malaysia_certified, EntryType.InEvaluation: get_malaysia_in_evaluation},
- "NL": {EntryType.Certified: get_netherlands_certified, EntryType.InEvaluation: get_netherlands_in_evaluation},
- "NO": {EntryType.Certified: get_norway_certified, EntryType.Archived: get_norway_archived},
- "KO": {EntryType.Certified: get_korea_certified, EntryType.Archived: get_korea_archived},
+ "MY": {
+ EntryType.Certified: get_malaysia_certified,
+ EntryType.Archived: get_malaysia_archived,
+ EntryType.InEvaluation: get_malaysia_in_evaluation,
+ },
+ "NL": {
+ EntryType.Certified: get_netherlands_certified,
+ EntryType.InEvaluation: get_netherlands_in_evaluation,
+ },
+ "NO": {
+ EntryType.Certified: get_norway_certified,
+ EntryType.Archived: get_norway_archived,
+ },
+ "KO": {
+ EntryType.Certified: get_korea_certified,
+ EntryType.Archived: get_korea_archived,
+ },
"SG": {
EntryType.InEvaluation: get_singapore_in_evaluation,
EntryType.Certified: get_singapore_certified,
diff --git a/src/sec_certs/sample/cve.py b/src/sec_certs/sample/cve.py
index 7f1a7cba..e1df1dbe 100644
--- a/src/sec_certs/sample/cve.py
+++ b/src/sec_certs/sample/cve.py
@@ -195,7 +195,7 @@ class CVE(PandasSerializableType, ComplexSerializableType):
@staticmethod
def parse_single_configuration(
- configuration: dict[str, Any]
+ configuration: dict[str, Any],
) -> tuple[list[CPEMatchCriteria], CPEMatchCriteriaConfiguration | None]:
if CVE.configuration_is_simple(configuration):
return CVE.get_simple_criteria_from_cpe_matches(configuration["nodes"][0]["cpeMatch"]), None
diff --git a/src/sec_certs/sample/fips_mip.py b/src/sec_certs/sample/fips_mip.py
index 0720e520..71868954 100644
--- a/src/sec_certs/sample/fips_mip.py
+++ b/src/sec_certs/sample/fips_mip.py
@@ -23,6 +23,7 @@ logger = logging.getLogger(__name__)
@total_ordering
class MIPStatus(Enum):
+ ON_HOLD = "On Hold"
IN_REVIEW = "In Review"
REVIEW_PENDING = "Review Pending"
COORDINATION = "Coordination"
diff --git a/src/sec_certs/sample/sar.py b/src/sec_certs/sample/sar.py
index e3d7bd2b..ea4a6e14 100644
--- a/src/sec_certs/sample/sar.py
+++ b/src/sec_certs/sample/sar.py
@@ -15,7 +15,7 @@ SAR_CLASS_MAPPING = {
"ALC": "Life-cycle support",
"ATE": "Tests",
"AVA": "Vulnerability assessment",
- "ACO": "Comoposition",
+ "ACO": "Composition",
}
SAR_CLASSES = set(SAR_CLASS_MAPPING)
diff --git a/src/sec_certs/utils/extract.py b/src/sec_certs/utils/extract.py
index e6312b8d..669cbcfc 100644
--- a/src/sec_certs/utils/extract.py
+++ b/src/sec_certs/utils/extract.py
@@ -818,3 +818,12 @@ def get_sums_for_rules_subset(dct: dict | None, path: str) -> dict[str, float]:
cc_rules_subset_to_search = rules_get_subset(path)
paths_to_search = extract_key_paths(cc_rules_subset_to_search, path)
return {x: get_sum_of_values_from_dict_path(dct, x, np.nan) for x in paths_to_search}
+
+
+scheme_frontpage_functions = {
+ "FR": search_only_headers_anssi,
+ "DE": search_only_headers_bsi,
+ "NL": search_only_headers_nscib,
+ "US": search_only_headers_niap,
+ "CA": search_only_headers_canada,
+}
generated by cgit v1.3.1 (git 2.54.0) at 2026-08-29 15:06:50 +0000