Certificate Manager

def delete(self, party_id) -> List[int]:
    """Deletes given party from table

    Args:
        party_id: Party id

    Returns:
        The document IDs of deleted certificates
    """

    return self._db.remove(self._query.party_id == party_id)

@staticmethod
def generate_self_signed_ssl_certificate(
    certificate_folder,
    certificate_name: str = "FBM_",
    component_id: str = "unknown",
    subject: Optional[Dict[str, str]] = None,
) -> Tuple[str, str]:
    """Creates self-signed certificates

    Args:
        certificate_folder: The path where certificate files `.pem` and `.key`
            will be saved. Path should be absolute.
        certificate_name: Name of the certificate file.
        component_id: ID of the component

    Returns:
        private_key: Private key file
        public_key: Certificate file

    Raises:
        FedbiomedCertificateError: If certificate directory is invalid or an error
            occurs while writing certificate files in given path.

    !!! info "Certificate files"
            Certificate files will be saved in the given directory as
            `certificates.key` for private key `certificate.pem` for public key.
    """
    subject = subject or {}

    if not os.path.abspath(certificate_folder):
        raise FedbiomedCertificateError(
            f"{ErrorNumbers.FB619.value}: Certificate path should be absolute: "
            f"{certificate_folder}"
        )

    if not os.path.isdir(certificate_folder):
        raise FedbiomedCertificateError(
            f"{ErrorNumbers.FB619.value}: Certificate path is not valid: {certificate_folder}"
        )

    pkey = crypto.PKey()
    pkey.generate_key(crypto.TYPE_RSA, 2048)

    cn = subject.get("CommonName", "*")
    on = subject.get("OrganizationName", component_id)

    x509 = crypto.X509()
    subject = x509.get_subject()
    subject.commonName = cn
    subject.organizationName = on
    x509.set_issuer(subject)

    extensions = []
    try:
        if ipaddress.ip_address(cn):
            extensions.append(
                crypto.X509Extension(
                    type_name=b"subjectAltName",
                    critical=False,
                    value=f"IP:{cn}".encode("ASCII"),
                )
            )
    except ValueError:
        pass
    if extensions:
        x509.add_extensions(extensions)

    x509.gmtime_adj_notBefore(0)
    x509.gmtime_adj_notAfter(5 * 365 * 24 * 60 * 60)
    x509.set_pubkey(pkey)
    x509.set_serial_number(random.randrange(100000))
    x509.set_version(2)
    x509.sign(pkey, "SHA256")

    # Certificate names
    key_file = os.path.join(certificate_folder, f"{certificate_name}.key")
    pem_file = os.path.join(certificate_folder, f"{certificate_name}.pem")

    try:
        with open(key_file, "wb") as f:
            f.write(crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey))
            f.close()
    except Exception as e:
        raise FedbiomedCertificateError(
            f"{ErrorNumbers.FB619.value}: Can not write public key: {e}"
        ) from e

    try:
        with open(pem_file, "wb") as f:
            f.write(crypto.dump_certificate(crypto.FILETYPE_PEM, x509))
            f.close()
    except Exception as e:
        raise FedbiomedCertificateError(
            f"{ErrorNumbers.FB619.value}: Can not write public key: {e}"
        ) from e

    return key_file, pem_file

def get(self, party_id: str) -> Union[dict, None]:
    """Gets certificate/public key  of given party

    Args:
        party_id: ID of the party which certificate will be retrieved from DB

    Returns:
        Certificate, dict like TinyDB document
    """

    v = self._db.get(self._query.party_id == party_id)
    return v

def insert(
    self,
    certificate: str,
    party_id: str,
    component: str,
    upsert: bool = False,
) -> Union[int, list[int]]:
    """Inserts new certificate

    Args:
        certificate: Public-key for the FL parties
        party_id: ID of the party
        component: Node or researcher,
        ip: IP of the component which the certificate will be registered
        port: Port of the component which the certificate will be registered
        upsert: Update document with new data if it is existing

    Returns:
        Document ID of inserted certificate

    Raises:
        FedbiomedCertificateError: party is already registered
    """
    certificate_ = self.get(party_id=party_id)
    if not certificate_:
        return self._db.insert(
            {
                "certificate": certificate,
                "party_id": party_id,
                "component": component,
            }
        )

    if upsert:
        return self._db.upsert(
            {
                "certificate": certificate,
                "component": component,
                "party_id": party_id,
            },
            self._query.party_id == party_id,
        )
    raise FedbiomedCertificateError(
        f"{ErrorNumbers.FB619.value}: Party {party_id} already registered. "
        "Please use `upsert=True` or '--upsert' option through CLI"
    )

def list(self, verbose: bool = False) -> List[dict]:
    """Lists registered certificates.

    Args:
        verbose: Prints list of registered certificates in tabular format

    Returns:
        List of certificate objects registered in DB
    """
    certificates = self._db.all()

    if verbose:
        to_print = copy.deepcopy(certificates)
        for doc in to_print:
            doc.pop("certificate")

        print(tabulate(to_print, headers="keys"))

    return certificates

def register_certificate(
    self,
    certificate_path: str,
    party_id: str,
    upsert: bool = False,
) -> Union[int, List[int]]:
    """Registers certificate

    Args:
        certificate_path: Path where certificate/key file stored
        party_id: ID of the FL party which the certificate will be registered
        upsert: If `True` overwrites existing certificate for specified party.  If `False`
            and the certificate for the specified party already existing it raises error.

    Returns:
        The document ID of registered certificated.

    Raises:
        FedbiomedCertificateError: If certificate file is not existing in file system
    """

    if not os.path.isfile(certificate_path):
        raise FedbiomedCertificateError(
            f"{ErrorNumbers.FB619.value}: Certificate path does not represents a file."
        )

    # Read certificate content
    certificate_content = read_file(certificate_path)

    # Save certificate in database
    component = (
        ComponentType.NODE.name
        if party_id.startswith(NODE_PREFIX)
        else ComponentType.RESEARCHER.name
    )

    return self.insert(
        certificate=certificate_content,
        party_id=party_id,
        component=component,
        upsert=upsert,
    )

def set_db(self, db_path: str) -> None:
    """Sets database

    Args:
        db_path: The path of DB file where `Certificates` table are stored
    """
    db = TinyDB(db_path)
    db.table_class = DBTable
    self._db: Table = db.table("Certificates")