Merge pull request #247 from crocs-muni/feat/scheme-pages

Add download of certificate lists from scheme pages.

3 files changed, 893 insertions, 1 deletions

diff --git a/sec_certs/constants.py b/sec_certs/constants.py
index 72b12d08..f6da500c 100644
--- a/sec_certs/constants.py
+++ b/sec_certs/constants.py
@@ -57,3 +57,51 @@ GARBAGE_SIZE_THRESHOLD = 1000
 GARBAGE_AVG_LLEN_THRESHOLD = 10
 GARBAGE_EVERY_SECOND_CHAR_THRESHOLD = 15
 GARBAGE_ALPHA_CHARS_THRESHOLD = 0.5
+
+CC_AUSTRALIA_BASE_URL = "https://www.cyber.gov.au"
+CC_AUSTRALIA_CERTIFIED_URL = (
+    CC_AUSTRALIA_BASE_URL + "/acsc/view-all-content/programs/australian-information-security-evaluation-program"
+)
+CC_CANADA_CERTIFIED_URL = "https://www.cyber.gc.ca/en/tools-services/common-criteria/certified-products"
+CC_CANADA_INEVAL_URL = "https://www.cyber.gc.ca/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_BSI_BASE_URL = "https://www.bsi.bund.de/"
+CC_BSI_CERTIFIED_URL = CC_BSI_BASE_URL + "EN/Topics/Certification/certified_products/certified_products_node.html"
+CC_INDIA_CERTIFIED_URL = "https://www.commoncriteria-india.gov.in/product-certified"
+CC_INDIA_ARCHIVED_URL = "https://www.commoncriteria-india.gov.in/archived-prod-cer"
+CC_ITALY_BASE_URL = "https://ocsi.isticom.it"
+CC_ITALY_CERTIFIED_URL = CC_ITALY_BASE_URL + "/index.php/elenchi-certificazioni/prodotti-certificati"
+CC_ITALY_INEVAL_URL = CC_ITALY_BASE_URL + "/index.php/elenchi-certificazioni/in-corso-di-valutazione"
+CC_JAPAN_CERTIFIED_URL = "https://www.ipa.go.jp/security/jisec/jisec_e/certified_products/certfy_list_e31.html"
+CC_JAPAN_ARCHIVED_URL = "https://www.ipa.go.jp/security/jisec/jisec_e/certified_products/certfy_list_e_archive.html"
+CC_JAPAN_INEVAL_URL = "https://www.ipa.go.jp/security/jisec/jisec_e/prdct_in_eval.html"
+CC_MALAYSIA_BASE_URL = "https://www.cybersecurity.my/mycc"
+CC_MALAYSIA_CERTIFIED_URL = CC_MALAYSIA_BASE_URL + "/mycprA.html"
+CC_MALAYSIA_INEVAL_URL = CC_MALAYSIA_BASE_URL + "/mycprC.html"
+CC_NETHERLANDS_BASE_URL = "https://www.tuv-nederland.nl/common-criteria"
+CC_NETHERLANDS_CERTIFIED_URL = CC_NETHERLANDS_BASE_URL + "/certificates.html"
+CC_NETHERLANDS_INEVAL_URL = CC_NETHERLANDS_BASE_URL + "/ongoing-certifications.html"
+CC_NORWAY_CERTIFIED_URL = "https://sertit.no/certified-products/category1919.html"
+CC_NORWAY_ARCHIVED_URL = "https://sertit.no/certified-products/product-archive/"
+CC_KOREA_EN_URL = "https://itscc.kr/main/main.do?accessMode=home_en"
+CC_KOREA_CERTIFIED_URL = "https://itscc.kr/certprod/list.do"
+CC_KOREA_PRODUCT_URL = "https://itscc.kr/certprod/view.do?product_id={}&product_class=1"
+CC_SINGAPORE_BASE_URL = "https://www.csa.gov.sg"
+CC_SINGAPORE_CERTIFIED_URL = (
+    CC_SINGAPORE_BASE_URL + "/Programmes/certification-and-labelling-schemes/csa-common-criteria/product-list"
+)
+CC_SINGAPORE_ARCHIVED_URL = (
+    CC_SINGAPORE_BASE_URL + "/Programmes/certification-and-labelling-schemes/csa-common-criteria/product-archives"
+)
+CC_SPAIN_BASE_URL = "https://oc.ccn.cni.es"
+CC_SPAIN_CERTIFIED_URL = CC_SPAIN_BASE_URL + "/en/certified-products/certified-products"
+CC_SWEDEN_BASE_URL = "https://www.fmv.se"
+CC_SWEDEN_CERTIFIED_URL = CC_SWEDEN_BASE_URL + "/verksamhet/ovrig-verksamhet/csec/certifikat-utgivna-av-csec/"
+CC_SWEDEN_INEVAL_URL = CC_SWEDEN_BASE_URL + "/verksamhet/ovrig-verksamhet/csec/pagaende-certifieringar/"
+CC_SWEDEN_ARCHIVED_URL = CC_SWEDEN_BASE_URL + "/verksamhet/ovrig-verksamhet/csec/arkiverade-certifikat-aldre-an-5-ar/"
+CC_TURKEY_ARCHIVED_URL = "https://statik.tse.org.tr/upload/tr/dosya/icerikyonetimi/3300/03112021143434-2.pdf"
+CC_USA_BASE_URL = "https://www.niap-ccevs.org"
+CC_USA_CERTIFIED_URL = CC_USA_BASE_URL + "/Product/PCL.cfm"
+CC_USA_INEVAL_URL = CC_USA_BASE_URL + "/Product/PINE.cfm"
+CC_USA_ARCHIVED_URL = CC_USA_BASE_URL + "/Product/Archived.cfm"
diff --git a/sec_certs/dataset/common_criteria.py b/sec_certs/dataset/common_criteria.py
index dfeb28bf..24d8bc41 100644
--- a/sec_certs/dataset/common_criteria.py
+++ b/sec_certs/dataset/common_criteria.py
@@ -13,9 +13,12 @@ from typing import Any, Callable, ClassVar, Dict, Iterator, List, Optional, Set,
 
 import numpy as np
 import pandas as pd
-from bs4 import BeautifulSoup, Tag
+import requests
+import tabula
+from bs4 import BeautifulSoup, NavigableString, Tag
 
 import sec_certs.utils.sanitization
+from sec_certs import constants
 from sec_certs.config.configuration import config
 from sec_certs.dataset.dataset import Dataset, logger
 from sec_certs.dataset.protection_profile import ProtectionProfileDataset
@@ -991,3 +994,772 @@ class CCDatasetMaintenanceUpdates(CCDataset, ComplexSerializableType):
         main_dates.maintenance_date = main_dates.maintenance_date.map(lambda x: [x])
         main_dates.index.name = "dgst"
         return main_dates.groupby("related_cert_digest").maintenance_date.agg("sum").rename("maintenance_dates")
+
+
+class CCSchemeDataset:
+    @staticmethod
+    def _download_page(url, session=None):
+        if session:
+            conn = session
+        else:
+            conn = requests
+        resp = conn.get(url, headers={"User-Agent": "seccerts.org"})
+        if resp.status_code != requests.codes.ok:
+            raise ValueError(f"Unable to download: status={resp.status_code}")
+        return BeautifulSoup(resp.content, "html.parser")
+
+    @staticmethod
+    def get_australia_in_evaluation():
+        # TODO: Information could be expanded by following url.
+        soup = CCSchemeDataset._download_page(constants.CC_AUSTRALIA_CERTIFIED_URL)
+        header = soup.find("h2", text="Products in evaluation")
+        table = header.find_next_sibling("table")
+        results = []
+        for tr in table.find_all("tr"):
+            tds = tr.find_all("td")
+            if not tds:
+                continue
+            cert = {
+                "vendor": str(tds[0].text),
+                "product": str(tds[1].text),
+                "url": constants.CC_AUSTRALIA_BASE_URL + str(tds[1].find("a")["href"]),
+                "level": str(tds[2].text),
+            }
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_canada_certified():
+        soup = CCSchemeDataset._download_page(constants.CC_CANADA_CERTIFIED_URL)
+        tbody = soup.find("table").find("tbody")
+        results = []
+        for tr in tbody.find_all("tr"):
+            tds = tr.find_all("td")
+            if not tds:
+                continue
+            cert = {
+                "product": str(tds[0].text),
+                "vendor": str(tds[1].text),
+                "level": str(tds[2].text),
+                "certification_date": str(tds[3].text),
+            }
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_canada_in_evaluation():
+        soup = CCSchemeDataset._download_page(constants.CC_CANADA_INEVAL_URL)
+        tbody = soup.find("table").find("tbody")
+        results = []
+        for tr in tbody.find_all("tr"):
+            tds = tr.find_all("td")
+            if not tds:
+                continue
+            cert = {
+                "product": str(tds[0].text),
+                "vendor": str(tds[1].text),
+                "level": str(tds[2].text),
+                "cert_lab": str(tds[3].text),
+            }
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_france_certified():
+        # TODO: Information could be expanded by following product link.
+        base_soup = CCSchemeDataset._download_page(constants.CC_ANSSI_CERTIFIED_URL)
+        category_nav = base_soup.find("ul", class_="nav-categories")
+        results = []
+        for li in category_nav.find_all("li"):
+            a = li.find("a")
+            url = a["href"]
+            category_name = str(a.text)
+            soup = CCSchemeDataset._download_page(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 tbody.find_all("tr"):
+                tds = tr.find_all("td")
+                if not tds:
+                    continue
+                cert = {
+                    "product": str(tds[0].text),
+                    "vendor": str(tds[1].text),
+                    "level": str(tds[2].text),
+                    "id": str(tds[3].text),
+                    "certification_date": str(tds[4].text),
+                    "category": category_name,
+                    "url": constants.CC_ANSSI_BASE_URL + str(tds[0].find("a")["href"]),
+                }
+                results.append(cert)
+        return results
+
+    @staticmethod
+    def get_germany_certified():
+        # TODO: Information could be expanded by following url.
+        base_soup = CCSchemeDataset._download_page(constants.CC_BSI_CERTIFIED_URL)
+        category_nav = base_soup.find("ul", class_="no-bullet row")
+        results = []
+        for li in category_nav.find_all("li"):
+            a = li.find("a")
+            url = a["href"]
+            category_name = str(a.text)
+            soup = CCSchemeDataset._download_page(constants.CC_BSI_BASE_URL + url)
+            content = soup.find("div", class_="content").find("div", class_="column")
+            for table in content.find_all("table"):
+                tbody = table.find("tbody")
+                header = table.find_parent("div", class_="wrapperTable").find_previous_sibling("h2")
+                for tr in tbody.find_all("tr"):
+                    tds = tr.find_all("td")
+                    if len(tds) != 4:
+                        continue
+                    cert = {
+                        "cert_id": str(tds[0].text),
+                        "product": str(tds[1].text),
+                        "vendor": str(tds[2].text),
+                        "certification_date": str(tds[3].text),
+                        "category": category_name,
+                        "url": constants.CC_BSI_BASE_URL + str(tds[0].find("a")["href"]),
+                    }
+                    if header is not None:
+                        cert["subcategory"] = str(header.text)
+                    results.append(cert)
+        return results
+
+    @staticmethod
+    def get_india_certified():
+        pages = {0}
+        seen_pages = set()
+        results = []
+        while pages:
+            page = pages.pop()
+            seen_pages.add(page)
+            url = constants.CC_INDIA_CERTIFIED_URL + f"?page={page}"
+            soup = CCSchemeDataset._download_page(url)
+
+            # Update pages
+            pager = soup.find("ul", class_="pager")
+            for li in pager.find_all("li"):
+                try:
+                    new_page = int(li.text)
+                except Exception:
+                    continue
+                if new_page not in seen_pages:
+                    pages.add(new_page)
+
+            # Parse table
+            tbody = soup.find("div", class_="content").find("table").find("tbody")
+            for tr in tbody.find_all("tr"):
+                tds = tr.find_all("td")
+                if not tds:
+                    continue
+                report_a = tds[5].find("a")
+                target_a = tds[6].find("a")
+                cert_a = tds[7].find("a")
+                cert = {
+                    "serial_number": str(tds[0].text),
+                    "product": str(tds[1].text),
+                    "sponsor": str(tds[2].text),
+                    "developer": str(tds[3].text),
+                    "level": str(tds[4].text),
+                    "report_link": report_a["href"],
+                    "report_name": str(report_a.text),
+                    "target_link": target_a["href"],
+                    "target_name": str(target_a.text),
+                    "cert_link": cert_a["href"],
+                    "cert_name": str(cert_a.text),
+                }
+                results.append(cert)
+        return results
+
+    @staticmethod
+    def get_india_archived():
+        pages = {0}
+        seen_pages = set()
+        results = []
+        while pages:
+            page = pages.pop()
+            seen_pages.add(page)
+            url = constants.CC_INDIA_ARCHIVED_URL + f"?page={page}"
+            soup = CCSchemeDataset._download_page(url)
+
+            # Update pages
+            pager = soup.find("ul", class_="pager")
+            for li in pager.find_all("li"):
+                try:
+                    new_page = int(li.text)
+                except Exception:
+                    continue
+                if new_page not in seen_pages:
+                    pages.add(new_page)
+
+            # Parse table
+            tbody = soup.find("div", class_="content").find("table").find("tbody")
+            for tr in tbody.find_all("tr"):
+                tds = tr.find_all("td")
+                if not tds:
+                    continue
+                report_a = tds[5].find("a")
+                target_a = tds[6].find("a")
+                cert_a = tds[7].find("a")
+                cert = {
+                    "serial_number": str(tds[0].text),
+                    "product": str(tds[1].text),
+                    "sponsor": str(tds[2].text),
+                    "developer": str(tds[3].text),
+                    "level": str(tds[4].text),
+                    "report_link": report_a["href"],
+                    "report_name": str(report_a.text),
+                    "target_link": target_a["href"],
+                    "target_name": str(target_a.text),
+                    "cert_link": cert_a["href"],
+                    "cert_name": str(cert_a.text),
+                    "certification_date": str(tds[8].text),
+                }
+                results.append(cert)
+        return results
+
+    @staticmethod
+    def get_italy_certified():  # noqa: C901
+        soup = CCSchemeDataset._download_page(constants.CC_ITALY_CERTIFIED_URL)
+        div = soup.find("div", class_="certificati")
+        results = []
+        for cert_div in div.find_all("div", recursive=False):
+            title = cert_div.find("h3").text
+            data_div = cert_div.find("div", class_="collapse")
+            cert = {"title": title}
+            for data_p in data_div.find_all("p"):
+                p_text = data_p.text
+                if ":" not in p_text:
+                    continue
+                p_name, p_data = p_text.split(":")
+                p_data = p_data.strip()
+                p_link = data_p.find("a")
+                if "Fornitore" in p_name:
+                    cert["supplier"] = p_data
+                elif "Livello di garanzia" in p_name:
+                    cert["level"] = p_data
+                elif "Data emissione certificato" in p_name:
+                    cert["certification_date"] = p_data
+                elif "Data revisione" in p_name:
+                    cert["revision_date"] = p_data
+                elif "Rapporto di Certificazione" in p_name and p_link:
+                    cert["report_link_it"] = constants.CC_ITALY_BASE_URL + p_link["href"]
+                elif "Certification Report" in p_name and p_link:
+                    cert["report_link_en"] = constants.CC_ITALY_BASE_URL + p_link["href"]
+                elif "Traguardo di Sicurezza" in p_name and p_link:
+                    cert["target_link"] = constants.CC_ITALY_BASE_URL + p_link["href"]
+                elif "Nota su" in p_name and p_link:
+                    cert["vulnerability_note_link"] = constants.CC_ITALY_BASE_URL + p_link["href"]
+                elif "Nota di chiarimento" in p_name and p_link:
+                    cert["clarification_note_link"] = constants.CC_ITALY_BASE_URL + p_link["href"]
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_italy_in_evaluation():
+        soup = CCSchemeDataset._download_page(constants.CC_ITALY_INEVAL_URL)
+        div = soup.find("div", class_="valutazioni")
+        results = []
+        for cert_div in div.find_all("div", recursive=False):
+            title = cert_div.find("h3").text
+            data_div = cert_div.find("div", class_="collapse")
+            cert = {"title": title}
+            for data_p in data_div.find_all("p"):
+                p_text = data_p.text
+                if ":" not in p_text:
+                    continue
+                p_name, p_data = p_text.split(":")
+                p_data = p_data.strip()
+                if "Committente" in p_name:
+                    cert["client"] = p_data
+                elif "Livello di garanzia" in p_name:
+                    cert["level"] = p_data
+                elif "Tipologia prodotto" in p_name:
+                    cert["product_type"] = p_data
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_japan_certified():
+        # TODO: Information could be expanded by following toe link.
+        soup = CCSchemeDataset._download_page(constants.CC_JAPAN_CERTIFIED_URL)
+        table = soup.find("div", id="cert_list").find("table")
+        results = []
+        trs = list(table.find_all("tr"))
+        for tr in trs:
+            tds = tr.find_all("td")
+            if not tds:
+                continue
+            if len(tds) == 6:
+                cert = {
+                    "cert_id": str(tds[0].text),
+                    "supplier": str(tds[1].text),
+                    "toe_overseas_name": str(tds[2].text),
+                    "certification_date": str(tds[3].text),
+                    "claim": str(tds[4].text),
+                }
+                toe_a = tds[2].find("a")
+                if toe_a and "href" in toe_a.attrs:
+                    cert["toe_overseas_link"] = toe_a["href"]
+                results.append(cert)
+            if len(tds) == 1:
+                cert = results[-1]
+                cert["toe_japan_name"] = str(tds[0].text)
+                toe_a = tds[0].find("a")
+                if toe_a and "href" in toe_a.attrs:
+                    cert["toe_japan_link"] = toe_a["href"]
+        return results
+
+    @staticmethod
+    def get_japan_archived():
+        # TODO: Information could be expanded by following toe link.
+        soup = CCSchemeDataset._download_page(constants.CC_JAPAN_ARCHIVED_URL)
+        table = soup.find("table")
+        results = []
+        trs = list(table.find_all("tr"))
+        for tr in trs:
+            tds = tr.find_all("td")
+            if not tds:
+                continue
+            if len(tds) == 6:
+                cert = {
+                    "cert_id": str(tds[0].text),
+                    "supplier": str(tds[1].text),
+                    "toe_overseas_name": str(tds[2].text),
+                    "certification_date": str(tds[3].text),
+                    "claim": str(tds[4].text),
+                }
+                toe_a = tds[2].find("a")
+                if toe_a and "href" in toe_a.attrs:
+                    cert["toe_overseas_link"] = toe_a["href"]
+                results.append(cert)
+            if len(tds) == 1:
+                cert = results[-1]
+                cert["toe_japan_name"] = str(tds[0].text)
+                toe_a = tds[0].find("a")
+                if toe_a and "href" in toe_a.attrs:
+                    cert["toe_japan_link"] = toe_a["href"]
+        return results
+
+    @staticmethod
+    def get_japan_in_evaluation():
+        # TODO: Information could be expanded by following toe link.
+        soup = CCSchemeDataset._download_page(constants.CC_JAPAN_INEVAL_URL)
+        table = soup.find("table")
+        results = []
+        for tr in table.find_all("tr"):
+            tds = tr.find_all("td")
+            if not tds:
+                continue
+            toe_a = tds[1].find("a")
+            cert = {
+                "supplier": str(tds[0].text),
+                "toe_name": str(toe_a.text),
+                "toe_link": toe_a["href"],
+                "claim": str(tds[2].text),
+            }
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_malaysia_certified():
+        soup = CCSchemeDataset._download_page(constants.CC_MALAYSIA_CERTIFIED_URL)
+        main_table = soup.find("table").find("table")
+        cert_table = main_table.find_all("table")[1].find("table")
+        tables = cert_table.find_all(lambda x: x.name == "table" and x.find_parent("table") == cert_table)
+        category_name = None
+        archive = False
+        results = []
+        for table in tables:
+            if table.find("table") is None:
+                name_td = table.find_all("td")[1]
+                if "Archive" in name_td.text:
+                    category_name = str(name_td.text).split("Archive")[1]
+                    archive = True
+                else:
+                    category_name = str(name_td.text)
+                    archive = False
+                continue
+            data_table = table.find("table")
+            tr = data_table.find_all("tr")[1]
+            tds = tr.find_all("td")
+            if len(tds) != 6:
+                continue
+            cert = {
+                "category": category_name,
+                "archived": archive,
+                "level": str(tds[0].text),
+                "cert_id": str(tds[1].text),
+                "certification_date": str(tds[2].text),
+                "product": str(tds[3].text),
+                "developer": str(tds[4].text),
+            }
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_malaysia_in_evalution():
+        soup = CCSchemeDataset._download_page(constants.CC_MALAYSIA_INEVAL_URL)
+        cert_table = soup.find("table").find("table").find_all("table")[1]
+        tables = cert_table.find_all(lambda x: x.name == "table" and x.find_parent("table") == cert_table)
+        results = []
+        for table in tables:
+            cat_p = table.find_previous_sibling("p")
+            data_table = table.find("table")
+            tr = data_table.find_all("tr")[1]
+            tds = tr.find_all("td")
+            if len(tds) != 5:
+                continue
+            cert = {
+                "category": str(cat_p.text),
+                "level": str(tds[0].text),
+                "project_id": str(tds[1].text),
+                "toe_name": str(tds[2].text),
+                "developer": str(tds[3].text),
+                "expected_completion_date": str(tds[4].text),
+            }
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_netherlands_certified():
+        soup = CCSchemeDataset._download_page(constants.CC_NETHERLANDS_CERTIFIED_URL)
+        main_div = soup.select("body > main > div > div > div > div:nth-child(2) > div.col-lg-9 > div:nth-child(3)")[0]
+        rows = main_div.find_all("div", class_="row", recursive=False)
+        modals = main_div.find_all("div", class_="modal", recursive=False)
+        results = []
+        for row, modal in zip(rows, modals):
+            row_entries = row.find_all("a")
+            modal_trs = modal.find_all("tr")
+            cert = {
+                "manufacturer": str(row_entries[0].text),
+                "product": str(row_entries[1].text),
+                "scheme": str(row_entries[2].text),
+                "cert_id": str(row_entries[3].text),
+            }
+            for tr in modal_trs:
+                th_text = tr.find("th").text
+                td = tr.find("td")
+                if "Manufacturer website" in th_text:
+                    cert["manufacturer_link"] = td.find("a")["href"]
+                elif "Assurancelevel" in th_text:
+                    cert["level"] = str(td.text)
+                elif "Certificate" in th_text:
+                    cert["cert_link"] = constants.CC_NETHERLANDS_BASE_URL + td.find("a")["href"]
+                elif "Certificationreport" in th_text:
+                    cert["report_link"] = constants.CC_NETHERLANDS_BASE_URL + td.find("a")["href"]
+                elif "Securitytarget" in th_text:
+                    cert["target_link"] = constants.CC_NETHERLANDS_BASE_URL + td.find("a")["href"]
+                elif "Maintenance report" in th_text:
+                    cert["maintenance_link"] = constants.CC_NETHERLANDS_BASE_URL + td.find("a")["href"]
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_netherlands_in_evaluation():
+        soup = CCSchemeDataset._download_page(constants.CC_NETHERLANDS_INEVAL_URL)
+        table = soup.find("table")
+        results = []
+        for tr in table.find_all("tr")[1:]:
+            tds = tr.find_all("td")
+            cert = {
+                "developer": str(tds[0].text),
+                "product": str(tds[1].text),
+                "category": str(tds[2].text),
+                "level": str(tds[3].text),
+                "certification_id": str(tds[4].text),
+            }
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def _get_norway(url):
+        # TODO: Information could be expanded by following product link.
+        soup = CCSchemeDataset._download_page(url)
+        results = []
+        for tr in soup.find_all("tr", class_="certified-product"):
+            tds = tr.find_all("td")
+            cert = {
+                "product": str(tds[0].text).strip(),
+                "product_link": tds[0].find("a")["href"],
+                "category": str(tds[1].find("p", class_="value").text),
+                "developer": str(tds[2].find("p", class_="value").text),
+                "certification_date": str(tds[3].find("time").text),
+            }
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_norway_certified():
+        return CCSchemeDataset._get_norway(constants.CC_NORWAY_CERTIFIED_URL)
+
+    @staticmethod
+    def get_norway_archived():
+        return CCSchemeDataset._get_norway(constants.CC_NORWAY_ARCHIVED_URL)
+
+    @staticmethod
+    def _get_korea(product_class):
+        # TODO: Information could be expanded by following product link.
+        session = requests.session()
+        session.get(constants.CC_KOREA_EN_URL)
+        # Get base page
+        url = constants.CC_KOREA_CERTIFIED_URL + f"?product_class={product_class}"
+        soup = CCSchemeDataset._download_page(url, session=session)
+        seen_pages = set()
+        pages = {1}
+        results = []
+        while pages:
+            page = pages.pop()
+            csrf = soup.find("form", id="fm").find("input", attrs={"name": "csrf"})["value"]
+            resp = session.post(url, data={"csrf": csrf, "selectPage": page, "product_class": product_class})
+            soup = BeautifulSoup(resp.content, "html.parser")
+            tbody = soup.find("table", class_="cpl").find("tbody")
+            for tr in tbody.find_all("tr"):
+                tds = tr.find_all("td")
+                if len(tds) != 6:
+                    continue
+                link = tds[0].find("a")
+                id = link["id"].split("-")[1]
+                cert = {
+                    "product": str(tds[0].text),
+                    "cert_id": str(tds[1].text),
+                    "product_link": constants.CC_KOREA_PRODUCT_URL.format(id),
+                    "vendor": str(tds[2].text),
+                    "level": str(tds[3].text),
+                    "category": str(tds[4].text),
+                    "certification_date": str(tds[5].text),
+                }
+                results.append(cert)
+            seen_pages.add(page)
+            page_links = soup.find("div", class_="paginate").find_all("a", class_="number_off")
+            for page_link in page_links:
+                try:
+                    new_page = int(page_link.text)
+                    if new_page not in seen_pages:
+                        pages.add(new_page)
+                except Exception:
+                    pass
+        return results
+
+    @staticmethod
+    def get_korea_certified():
+        return CCSchemeDataset._get_korea(product_class=1)
+
+    @staticmethod
+    def get_korea_suspended():
+        return CCSchemeDataset._get_korea(product_class=2)
+
+    @staticmethod
+    def get_korea_archived():
+        return CCSchemeDataset._get_korea(product_class=4)
+
+    @staticmethod
+    def _get_singapore(url):
+        soup = CCSchemeDataset._download_page(url)
+        table = soup.find("table")
+        skip = False
+        results = []
+        category_name = None
+        for tr in table.find_all("tr"):
+            if skip:
+                skip = False
+                continue
+            tds = tr.find_all("td")
+            if len(tds) == 1:
+                category_name = str(tds[0].text)
+                skip = True
+                continue
+
+            cert = {
+                "product": str(tds[0].text.split()[0]),
+                "vendor": str(tds[1].text),
+                "level": str(tds[2].text),
+                "certification_date": str(tds[3].text),
+                "expiration_date": str(tds[4].text),
+                "category": category_name,
+            }
+            for link in tds[0].find_all("a"):
+                link_text = link.text.strip()
+                if link_text == "Certificate":
+                    cert["cert_link"] = constants.CC_SINGAPORE_BASE_URL + link["href"]
+                elif link_text in ("Certificate Report", "Certification Report"):
+                    cert["report_link"] = constants.CC_SINGAPORE_BASE_URL + link["href"]
+                elif link_text == "Security Target":
+                    cert["target_link"] = constants.CC_SINGAPORE_BASE_URL + link["href"]
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_singapore_certified():
+        return CCSchemeDataset._get_singapore(constants.CC_SINGAPORE_CERTIFIED_URL)
+
+    @staticmethod
+    def get_singapore_in_evaluation():
+        soup = CCSchemeDataset._download_page(constants.CC_SINGAPORE_CERTIFIED_URL)
+        header = soup.find(lambda x: x.name == "h3" and x.text == "In Evaluation")
+        table = header.find_next("table")
+        results = []
+        for tr in table.find_all("tr")[1:]:
+            tds = tr.find_all("td")
+            cert = {"name": str(tds[0].text), "vendor": str(tds[1].text), "level": str(tds[2].text)}
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_singapore_archived():
+        return CCSchemeDataset._get_singapore(constants.CC_SINGAPORE_ARCHIVED_URL)
+
+    @staticmethod
+    def get_spain_certified():
+        soup = CCSchemeDataset._download_page(constants.CC_SPAIN_CERTIFIED_URL)
+        tbody = soup.find("table", class_="djc_items_table").find("tbody")
+        results = []
+        for tr in tbody.find_all("tr", recursive=False):
+            tds = tr.find_all("td")
+            cert = {
+                "product": str(tds[0].text),
+                "product_link": constants.CC_SPAIN_BASE_URL + tds[0].find("a")["href"],
+                "category": str(tds[1].text),
+                "manufacturer": str(tds[2].text),
+                "certification_date": str(tds[3].text),
+            }
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def _get_sweden(url):
+        # TODO: Information could be expanded by following product link.
+        soup = CCSchemeDataset._download_page(url)
+        nav = soup.find("main").find("nav", class_="component-nav-box__list")
+        results = []
+        for link in nav.find_all("a"):
+            cert = {"product": str(link.text), "product_link": link["href"]}
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_sweden_certified():
+        return CCSchemeDataset._get_sweden(constants.CC_SWEDEN_CERTIFIED_URL)
+
+    @staticmethod
+    def get_sweden_in_evalution():
+        return CCSchemeDataset._get_sweden(constants.CC_SWEDEN_INEVAL_URL)
+
+    @staticmethod
+    def get_sweden_archived():
+        return CCSchemeDataset._get_sweden(constants.CC_SWEDEN_ARCHIVED_URL)
+
+    @staticmethod
+    def get_turkey_certified():
+        results = []
+        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}")
+            with pdf_path.open("wb") as f:
+                f.write(resp.content)
+            dfs = tabula.read_pdf(str(pdf_path), pages="all")
+            for df in dfs:
+                for line in df.values:
+                    cert = {
+                        # TODO: Split item number and generate several dicts for a range they include.
+                        "item_no": line[0],
+                        "developer": line[1],
+                        "product": line[2],
+                        "cc_version": line[3],
+                        "level": line[4],
+                        "cert_lab": line[5],
+                        "certification_date": line[6],
+                        "expiration_date": line[7],
+                        # TODO: Parse "Ongoing Evaluation" out of this field as well.
+                        "archived": isinstance(line[9], str) and "Archived" in line[9],
+                    }
+                    results.append(cert)
+        return results
+
+    @staticmethod
+    def get_usa_certified():
+        # TODO: Information could be expanded by following product link.
+        # TODO: Information could be expanded by following the cc_claims (has links to protection profiles).
+        soup = CCSchemeDataset._download_page(constants.CC_USA_CERTIFIED_URL)
+        tbody = soup.find("table", class_="tablesorter").find("tbody")
+        results = []
+        for tr in tbody.find_all("tr"):
+            tds = tr.find_all("td")
+            vendor_span = tds[0].find("span", class_="b u")
+            product_link = tds[0].find("a")
+            scheme_img = tds[6].find("img")
+            # Only return the US certifications.
+            if scheme_img["title"] != "USA":
+                continue
+            cert = {
+                "product": str(product_link.text),
+                "vendor": str(vendor_span.text),
+                "product_link": product_link["href"],
+                "id": str(tds[1].text),
+                "cc_claim": str(tds[2].text),
+                "cert_lab": str(tds[3].text),
+                "certification_date": str(tds[4].text),
+                "assurance_maintenance_date": str(tds[5].text),
+            }
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_usa_in_evaluation():
+        # TODO: Information could be expanded by following the cc_claims (has links to protection profiles).
+        soup = CCSchemeDataset._download_page(constants.CC_USA_INEVAL_URL)
+        tbody = soup.find("table", class_="tablesorter").find("tbody")
+        results = []
+        for tr in tbody.find_all("tr"):
+            tds = tr.find_all("td")
+            vendor_span = tds[0].find("span", class_="b u")
+            product_name = None
+            for child in tds[0].children:
+                if isinstance(child, NavigableString):
+                    product_name = str(child).strip()
+                    break
+            cert = {
+                "vendor": str(vendor_span.text),
+                "id": str(tds[1].text),
+                "cc_claim": str(tds[2].text),
+                "cert_lab": str(tds[3].text),
+                "kickoff_date": str(tds[4].text),
+            }
+            if product_name:
+                cert["product"] = product_name
+            results.append(cert)
+        return results
+
+    @staticmethod
+    def get_usa_archived():
+        # TODO: Information could be expanded by following the cc_claims (has links to protection profiles).
+        soup = CCSchemeDataset._download_page(constants.CC_USA_ARCHIVED_URL)
+        tbody = soup.find("table", class_="tablesorter").find("tbody")
+        results = []
+        for tr in tbody.find_all("tr"):
+            tds = tr.find_all("td")
+            scheme_img = tds[5].find("img")
+            # Only return the US certifications.
+            if scheme_img["title"] != "USA":
+                continue
+            vendor_span = tds[0].find("span", class_="b u")
+            product_name = None
+            for child in tds[0].children:
+                if isinstance(child, NavigableString):
+                    product_name = str(child).strip()
+                    break
+            cert = {
+                "vendor": str(vendor_span.text),
+                "id": str(tds[1].text),
+                "cc_claim": str(tds[2].text),
+                "cert_lab": str(tds[3].text),
+                "certification_date": str(tds[4].text),
+            }
+            if product_name:
+                cert["product"] = product_name
+            results.append(cert)
+        return results
diff --git a/tests/test_cc_schemes.py b/tests/test_cc_schemes.py
new file mode 100644
index 00000000..be840edb
--- /dev/null
+++ b/tests/test_cc_schemes.py
@@ -0,0 +1,72 @@
+from unittest import TestCase
+
+import pytest
+
+from sec_certs.dataset.common_criteria import CCSchemeDataset
+
+
+class TestCCSchemes(TestCase):
+    def test_australia(self):
+        assert len(CCSchemeDataset.get_australia_in_evaluation()) != 0
+
+    def test_canada(self):
+        assert len(CCSchemeDataset.get_canada_certified()) != 0
+        assert len(CCSchemeDataset.get_canada_in_evaluation()) != 0
+
+    def test_anssi(self):
+        assert len(CCSchemeDataset.get_france_certified()) != 0
+
+    def test_bsi(self):
+        assert len(CCSchemeDataset.get_germany_certified()) != 0
+
+    @pytest.mark.xfail
+    def test_india(self):
+        assert len(CCSchemeDataset.get_india_certified()) != 0
+        assert len(CCSchemeDataset.get_india_archived()) != 0
+
+    def test_italy(self):
+        assert len(CCSchemeDataset.get_italy_certified()) != 0
+        assert len(CCSchemeDataset.get_italy_in_evaluation()) != 0
+
+    def test_japan(self):
+        assert len(CCSchemeDataset.get_japan_certified()) != 0
+        assert len(CCSchemeDataset.get_japan_archived()) != 0
+        assert len(CCSchemeDataset.get_japan_in_evaluation()) != 0
+
+    def test_malaysia(self):
+        assert len(CCSchemeDataset.get_malaysia_certified()) != 0
+        assert len(CCSchemeDataset.get_malaysia_in_evalution()) != 0
+
+    def test_netherlands(self):
+        assert len(CCSchemeDataset.get_netherlands_certified()) != 0
+        assert len(CCSchemeDataset.get_netherlands_in_evaluation()) != 0
+
+    def test_norway(self):
+        assert len(CCSchemeDataset.get_norway_certified()) != 0
+        assert len(CCSchemeDataset.get_norway_archived()) != 0
+
+    def test_korea(self):
+        assert len(CCSchemeDataset.get_korea_certified()) != 0
+        CCSchemeDataset.get_korea_suspended()
+        assert len(CCSchemeDataset.get_korea_archived()) != 0
+
+    def test_singapore(self):
+        assert len(CCSchemeDataset.get_singapore_certified()) != 0
+        assert len(CCSchemeDataset.get_singapore_archived()) != 0
+        assert len(CCSchemeDataset.get_singapore_in_evaluation()) != 0
+
+    def test_spain(self):
+        assert len(CCSchemeDataset.get_spain_certified()) != 0
+
+    def test_sweden(self):
+        assert len(CCSchemeDataset.get_sweden_certified()) != 0
+        assert len(CCSchemeDataset.get_sweden_in_evalution()) != 0
+        assert len(CCSchemeDataset.get_sweden_archived()) != 0
+
+    def test_turkey(self):
+        assert len(CCSchemeDataset.get_turkey_certified()) != 0
+
+    def test_usa(self):
+        assert len(CCSchemeDataset.get_usa_certified()) != 0
+        assert len(CCSchemeDataset.get_usa_in_evaluation()) != 0
+        assert len(CCSchemeDataset.get_usa_archived()) != 0