Merge pull request #83 from crocs-muni/cc-serialization

Cc serialization

12 files changed, 55 insertions, 142 deletions

diff --git a/README.md b/README.md
index 255f726a..b6600896 100644
--- a/README.md
+++ b/README.md
@@ -13,7 +13,7 @@ This project is developed by the [Centre for Research On Cryptography and Securi
 
 ## Installation (CC)
 
-The tool requires `Python >=3.8` and `pdftotext` binary somewhere on the `PATH`.
+The tool requires `Python >=3.8` and `pdftotext` binary somewhere on the `PATH` ([More info about pdftotext dependencies](https://pypi.org/project/pdftotext/)).
 
 The stable release is published on [PyPi](https://pypi.org/project/sec-certs/) as well as on [DockerHub](https://hub.docker.com/repository/docker/seccerts/sec-certs), you can install it with:
 
@@ -68,7 +68,7 @@ dset = CCDataset.from_json('./latest_cc_snapshot.json')  # you can now load your
 If you wish to fully process the Common Criteria (CC) data by yourself, you can do that as follows. Running
 
 ```python
-cc-cli all --output ./cc_dataset
+cc-certs all --output ./cc_dataset
 ```
 
 will fully process the Common Criteria dataset, which can take up to 6 hours to finish. You can select only same tasks to run. Calling `cc-cli --help` yields
diff --git a/cc_cli.py b/cc_cli.py
index 41677882..41677882 100644..100755
--- a/cc_cli.py
+++ b/cc_cli.py
diff --git a/sec_certs/certificate/cc_maintenance_update.py b/sec_certs/certificate/cc_maintenance_update.py
index 4b9eb6d1..27cbec63 100644
--- a/sec_certs/certificate/cc_maintenance_update.py
+++ b/sec_certs/certificate/cc_maintenance_update.py
@@ -1,6 +1,7 @@
 import copy
 import logging
-from typing import Union, Optional, Set, Dict, ClassVar, List
+from abc import ABC
+from typing import Optional, Dict, List
 from datetime import date
 
 import sec_certs.helpers as helpers
@@ -11,9 +12,7 @@ from sec_certs.certificate.certificate import Certificate
 logger = logging.getLogger(__name__)
 
 
-class CommonCriteriaMaintenanceUpdate(CommonCriteriaCert):
-    relevant_fields: ClassVar[List[str]] = ['name', 'report_link', 'st_link', 'state', 'pdf_data', 'heuristics', 'related_cert_digest', 'maintenance_date']
-
+class CommonCriteriaMaintenanceUpdate(CommonCriteriaCert, ABC):
     def __init__(self, name: str, report_link: str, st_link: str,
                  state: Optional[CommonCriteriaCert.InternalState],
                  pdf_data: Optional[CommonCriteriaCert.PdfData],
@@ -27,6 +26,10 @@ class CommonCriteriaMaintenanceUpdate(CommonCriteriaCert):
         self.maintenance_date = maintenance_date
 
     @property
+    def serialized_attributes(self) -> List[str]:
+        return ['name', 'report_link', 'st_link', 'state', 'pdf_data', 'heuristics', 'related_cert_digest', 'maintenance_date']
+
+    @property
     def dgst(self):
         return 'cert_' + self.related_cert_digest + '_update_' + helpers.get_first_16_bytes_sha256(self.name)
 
@@ -34,9 +37,6 @@ class CommonCriteriaMaintenanceUpdate(CommonCriteriaCert):
     def from_dict(cls, dct: Dict) -> 'CommonCriteriaMaintenanceUpdate':
         return Certificate.from_dict(dct)
 
-    def to_dict(self):
-        return {key: val for key, val in copy.deepcopy(self.__dict__).items() if key in self.relevant_fields}
-
     @classmethod
     def get_updates_from_cc_cert(cls, cert: CommonCriteriaCert):
         return [cls(x.maintainance_title, x.maintainance_report_link, x.maintainance_st_link,
diff --git a/sec_certs/certificate/common_criteria.py b/sec_certs/certificate/common_criteria.py
index 9da17f0e..6ebde681 100644
--- a/sec_certs/certificate/common_criteria.py
+++ b/sec_certs/certificate/common_criteria.py
@@ -42,13 +42,6 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType):
                                 helpers.sanitize_string(self.maintainance_title))
             super().__setattr__('maintainance_date', helpers.sanitize_date(self.maintainance_date))
 
-        def to_dict(self):
-            return copy.deepcopy(self.__dict__)
-
-        @classmethod
-        def from_dict(cls, dct):
-            return cls(*tuple(dct.values()))
-
         def __lt__(self, other):
             return self.maintainance_date < other.maintainance_date
 
@@ -82,16 +75,6 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType):
             else:
                 self.errors = errors
 
-        def to_dict(self):
-            return {'st_link_ok': self.st_link_ok, 'report_link_ok': self.report_link_ok,
-                    'st_convert_ok': self.st_convert_ok, 'report_convert_ok': self.report_convert_ok,
-                    'st_extract_ok': self.st_extract_ok, 'report_extract_ok': self.report_extract_ok,
-                    'errors': self.errors}
-
-        @classmethod
-        def from_dict(cls, dct: Dict[str, bool]):
-            return cls(*tuple(dct.values()))
-
     @dataclass(init=False)
     class PdfData(ComplexSerializableType):
         report_metadata: Dict[str, Any]
@@ -115,12 +98,6 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType):
         def __bool__(self):
             return any([x is not None for x in vars(self)])
 
-        def to_dict(self):
-            return {'report_metadata': self.report_metadata, 'st_metadata': self.st_metadata,
-                    'report_frontpage': self.report_frontpage,
-                    'st_frontpage': self.st_frontpage, 'report_keywords': self.report_keywords,
-                    'st_keywords': self.st_keywords}
-
         @property
         def bsi_data(self) -> Optional[Dict[str, Any]]:
             return self.report_frontpage['bsi']
@@ -177,10 +154,6 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType):
         def cert_id(self) -> Optional[str]:
             return processed if (processed := self.processed_cert_id) else self.keywords_cert_id
 
-        @classmethod
-        def from_dict(cls, dct: Dict[str, bool]):
-            return cls(*tuple(dct.values()))
-
     @dataclass
     class Heuristics(ComplexSerializableType):
         extracted_versions: List[str] = field(default=None)
@@ -195,16 +168,15 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType):
 
         cpe_candidate_vendors: Optional[List[str]] = field(init=False)
 
+        @property
+        def serialized_attributes(self) -> List[str]:
+            all_vars = copy.deepcopy(super().serialized_attributes)
+            all_vars.remove('cpe_candidate_vendors')
+            return all_vars
+
         def __post_init__(self):
             self.cpe_candidate_vendors = None
 
-        def to_dict(self):
-            return {'extracted_versions': self.extracted_versions, 'cpe_matches': self.cpe_matches, 'labeled': self.labeled, 'verified_cpe_matches': self.verified_cpe_matches, 'related_cves': self.related_cves, 'cert_lab': self.cert_lab, 'cert_id': self.cert_id}
-
-        @classmethod
-        def from_dict(cls, dct: Dict[str, str]):
-            return cls(*tuple(dct.values()))
-
     pandas_columns = ['dgst', 'name', 'status', 'category', 'manufacturer', 'scheme', 'security_level',
                       'not_valid_before', 'not_valid_after', 'report_link', 'st_link',
                       'manufacturer_web', 'extracted_versions', 'cpe_matches', 'verified_cpe_matches',
diff --git a/sec_certs/certificate/fips.py b/sec_certs/certificate/fips.py
index 6ed7f856..80f9badc 100644
--- a/sec_certs/certificate/fips.py
+++ b/sec_certs/certificate/fips.py
@@ -27,15 +27,6 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
 
     @dataclass(eq=True)
     class State(ComplexSerializableType):
-
-        @classmethod
-        def from_dict(cls, dct: Dict):
-            return cls(Path(dct['sp_path']), Path(dct['html_path']), Path(dct['fragment_path']), dct['tables_done'],
-                       dct['file_status'], dct['txt_state'])
-
-        def to_dict(self):
-            return self.__dict__
-
         sp_path: Path
         html_path: Path
         fragment_path: Path
@@ -43,6 +34,15 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
         file_status: Optional[bool]
         txt_state: bool
 
+        def __init__(self, sp_path: Union[str, Path], html_path: Union[str, Path], fragment_path: Union[str, Path],
+                     tables_done: bool, file_status: Optional[bool], txt_state: bool):
+            self.sp_path = Path(sp_path)
+            self.html_path = Path(html_path)
+            self.fragment_path = Path(fragment_path)
+            self.tables_done = tables_done
+            self.file_status = file_status
+            self.txt_state = txt_state
+
     @dataclass(eq=True)
     class Algorithm(ComplexSerializableType):
         cert_id: str
@@ -63,14 +63,6 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
         def __str__(self):
             return str(self.type + ' algorithm #' + self.cert_id + ' created by ' + self.vendor)
 
-        def to_dict(self):
-            return copy.deepcopy(self.__dict__)
-
-        @classmethod
-        def from_dict(cls, dct: dict) -> 'FIPSCertificate.Algorithm':
-            return cls(dct['cert_id'], dct['vendor'], dct['implementation'], dct['type'],
-                       dct['date'])
-
     @dataclass(eq=True)
     class WebScan(ComplexSerializableType):
         module_name: Optional[str]
@@ -120,13 +112,6 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
         def __str__(self):
             return str(self.module_name + ' created by ' + self.vendor)
 
-        def to_dict(self):
-            return copy.deepcopy(self.__dict__)
-
-        @classmethod
-        def from_dict(cls, dct: dict) -> 'FIPSCertificate.WebScan':
-            return cls(*tuple(dct.values()))
-
     @dataclass(eq=True)
     class PdfScan(ComplexSerializableType):
         cert_id: int
@@ -146,13 +131,6 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
         def __str__(self):
             return str(self.cert_id)
 
-        def to_dict(self):
-            return copy.deepcopy(self.__dict__)
-
-        @classmethod
-        def from_dict(cls, dct: dict) -> 'FIPSCertificate.PdfScan':
-            return cls(*tuple(dct.values()))
-
     @dataclass(eq=True)
     class Processed(ComplexSerializableType):
         keywords: Optional[Dict[str, Dict]]
@@ -166,19 +144,9 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
             # for each id
             return helpers.get_first_16_bytes_sha256(str(self.keywords))
 
-        def to_dict(self):
-            return copy.deepcopy(self.__dict__)
-
-        @classmethod
-        def from_dict(cls, dct: dict) -> 'FIPSCertificate.Processed':
-            return cls(*tuple(dct.values()))
-
     def __str__(self) -> str:
         return str(self.cert_id)
 
-    def to_dict(self) -> Dict:
-        return self.__dict__
-
     @property
     def dgst(self) -> str:
         return self.cert_id
@@ -190,10 +158,6 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
             logger.error(
                 f'Failed to download security policy from {cert[0]}, code: {exit_code}')
 
-    @classmethod
-    def from_dict(cls, dct: dict):
-        return cls(*tuple(dct.values()))
-
     def __init__(self, cert_id: str,
                  web_scan: 'FIPSCertificate.WebScan',
                  pdf_scan: 'FIPSCertificate.PdfScan',
diff --git a/sec_certs/certificate/protection_profile.py b/sec_certs/certificate/protection_profile.py
index b18fde73..1bca2c31 100644
--- a/sec_certs/certificate/protection_profile.py
+++ b/sec_certs/certificate/protection_profile.py
@@ -23,9 +23,6 @@ class ProtectionProfile(ComplexSerializableType):
         super().__setattr__('pp_name', helpers.sanitize_string(self.pp_name))
         super().__setattr__('pp_link', helpers.sanitize_link(self.pp_link))
 
-    def to_dict(self):
-        return copy.deepcopy(self.__dict__)
-
     @classmethod
     def from_dict(cls, dct):
         new_dct = copy.deepcopy(dct)
diff --git a/sec_certs/dataset/common_criteria.py b/sec_certs/dataset/common_criteria.py
index 18db1941..3add738e 100644
--- a/sec_certs/dataset/common_criteria.py
+++ b/sec_certs/dataset/common_criteria.py
@@ -34,16 +34,9 @@ class CCDataset(Dataset, ComplexSerializableType):
         txt_data_extracted: bool = False
         certs_analyzed: bool = False
 
-        def to_dict(self):
-            return copy.deepcopy(self.__dict__)
-
         def __bool__(self):
             return any(vars(self))
 
-        @classmethod
-        def from_dict(cls, dct: Dict[str, bool]):
-            return cls(*tuple(dct.values()))
-
     certs: Dict[str, 'CommonCriteriaCert']
 
     def __init__(self, certs: Dict[str, 'CommonCriteriaCert'], root_dir: Path, name: str = 'dataset name',
diff --git a/sec_certs/dataset/cpe.py b/sec_certs/dataset/cpe.py
index eabda417..6cf9a3fb 100644
--- a/sec_certs/dataset/cpe.py
+++ b/sec_certs/dataset/cpe.py
@@ -36,12 +36,9 @@ class CPE(ComplexSerializableType):
             self.item_name = ' '.join(self.uri.split(':')[4].split('_'))
             self.version = self.uri.split(':')[5]
 
-    def to_dict(self):
-        return {'uri': self.uri, 'title': self.title}
-
-    @classmethod
-    def from_dict(cls, dct: Dict[str, str]):
-        return cls(*tuple(dct.values()))
+    @property
+    def serialized_attributes(self) -> List[str]:
+        return ['uri', 'title']
 
     def __hash__(self):
         return hash(self.uri)
diff --git a/sec_certs/dataset/cve.py b/sec_certs/dataset/cve.py
index d6cd3bb1..e2397690 100644
--- a/sec_certs/dataset/cve.py
+++ b/sec_certs/dataset/cve.py
@@ -28,13 +28,6 @@ class CVE(ComplexSerializableType):
         explotability_score: float
         impact_score: float
 
-        def to_dict(self):
-            return copy.deepcopy(self.__dict__)
-
-        @classmethod
-        def from_dict(cls, dct: Dict):
-            return cls(*tuple(dct.values()))
-
         @classmethod
         def from_nist_dict(cls, dct: Dict):
             """
@@ -57,13 +50,6 @@ class CVE(ComplexSerializableType):
     vulnerable_cpes: List[str]
     impact: Impact
 
-    def to_dict(self):
-        return copy.deepcopy(self.__dict__)
-
-    @classmethod
-    def from_dict(cls, dct: Dict):
-        return cls(*tuple(dct.values()))
-
     @classmethod
     def from_nist_dict(cls, dct: Dict) -> 'CVE':
         """
@@ -102,12 +88,9 @@ class CVEDataset(ComplexSerializableType):
 
     cve_url: ClassVar[str] = 'https://nvd.nist.gov/feeds/json/cve/1.1/nvdcve-1.1-'
 
-    def to_dict(self):
-        return copy.deepcopy({'cves': self.cves})
-
-    @classmethod
-    def from_dict(cls, dct: Dict):
-        return cls(*tuple(dct.values()))
+    @property
+    def serialized_attributes(self) -> List[str]:
+        return ['cves']
 
     def __post_init__(self):
         self.cpes_to_cve_lookup = dict()
diff --git a/sec_certs/dataset/fips_algorithm.py b/sec_certs/dataset/fips_algorithm.py
index 6194b12f..a9187816 100644
--- a/sec_certs/dataset/fips_algorithm.py
+++ b/sec_certs/dataset/fips_algorithm.py
@@ -1,7 +1,7 @@
 import json
 import logging
 from pathlib import Path
-from typing import Dict, Union
+from typing import Dict, Union, List
 
 from bs4 import BeautifulSoup
 
@@ -12,8 +12,8 @@ from sec_certs.serialization import ComplexSerializableType, CustomJSONEncoder,
 from sec_certs.certificate.fips import FIPSCertificate
 from sec_certs.configuration import config
 
-class FIPSAlgorithmDataset(Dataset, ComplexSerializableType):
 
+class FIPSAlgorithmDataset(Dataset, ComplexSerializableType):
     def get_certs_from_web(self):
         self.root_dir.mkdir(exist_ok=True)
         algs_paths, algs_urls = [], []
@@ -69,8 +69,9 @@ class FIPSAlgorithmDataset(Dataset, ComplexSerializableType):
     def download_all_pdfs(self):
         raise NotImplementedError('Not meant to be implemented')
 
-    def to_dict(self):
-        return {"certs": self.certs}
+    @property
+    def serialized_attributes(self) -> List[str]:
+        return ['certs']
 
     @classmethod
     def from_dict(cls, dct: Dict):
diff --git a/sec_certs/serialization.py b/sec_certs/serialization.py
index 80c20e4a..03ed00e9 100644
--- a/sec_certs/serialization.py
+++ b/sec_certs/serialization.py
@@ -1,21 +1,26 @@
 import json
 from datetime import date
 from pathlib import Path
-from typing import Dict
+from typing import Dict, List
+import copy
 
-from abc import ABC, abstractmethod
 
+class ComplexSerializableType:
+    # Ideally, the serialized_fields would be an class variable referencing itself, but that it virtually impossible
+    # to achieve without using metaclasses. Not to complicate the code, we choose instance variable.
+    @property
+    def serialized_attributes(self) -> List[str]:
+        return list(self.__dict__.keys())
 
-class ComplexSerializableType(ABC):
-    @classmethod
-    @abstractmethod
-    def to_dict(cls):
-        raise NotImplementedError
+    def __init__(self, *args):
+        pass
+
+    def to_dict(self):
+        return {key: val for key, val in copy.deepcopy(self.__dict__).items() if key in self.serialized_attributes}
 
     @classmethod
-    @abstractmethod
     def from_dict(cls, dct: Dict):
-        raise NotImplementedError
+        return cls(*(tuple(dct.values())))
 
 
 class CustomJSONEncoder(json.JSONEncoder):
diff --git a/setup.py b/setup.py
index c0b05e42..ce27cd0f 100644
--- a/setup.py
+++ b/setup.py
@@ -33,9 +33,10 @@ setup(
         "test": ["pytest", "coverage"]
     },
     include_package_data=True,
-    entry_points="""
-        [console_scripts]
-        cc-certs=cc_cli:main
-        fips-certs=sec_certs.entrypoints.fips_certificates:main
-    """
+    entry_points={
+        'console_scripts': [
+            'cc-certs=cc_cli:main',
+            'fips-certs=sec_certs.entrypoints.fips_certificates:main'
+        ]
+    }
 )