chore: move utils functions to init

pki_tools/__init__.py

@@ -1,5 +1,41 @@
 from . import ocsp
 from . import crl
 from . import exceptions
-from . import utils
 from . import types
+
+from typing import Union
+
+from cryptography import x509
+
+from loguru import logger
+
+
+def cert_from_pem(cert_pem: str) -> x509.Certificate:
+    try:
+        return x509.load_pem_x509_certificate(cert_pem.encode())
+    except ValueError as e:
+        raise exceptions.CertLoadError(e)
+
+
+def is_revoked(
+    cert: Union[x509.Certificate, types.PemCert],
+    issuer_cert: Union[
+        x509.Certificate, types.PemCert, types.OcspIssuerUri
+    ] = None,
+    crl_cache_seconds: int = 3600,
+) -> bool:
+    if issuer_cert is not None:
+        try:
+            return ocsp.is_revoked(cert, issuer_cert)
+        except exceptions.ExtensionMissing:
+            logger.debug("OCSP Extension missing, trying CRL next")
+
+    try:
+        return crl.is_revoked(cert, crl_cache_seconds)
+    except exceptions.ExtensionMissing:
+        err_msg = (
+            "OCSP and CRL extensions not found, "
+            "couldn't check revocation status"
+        )
+        logger.error(err_msg)
+        raise exceptions.Error(err_msg)

pki_tools/crl.py

@@ -8,16 +8,15 @@
 from cryptography.x509.oid import ExtensionOID
 from loguru import logger
 
-from pki_tools import exceptions
-from pki_tools import types
-from pki_tools import utils
+import pki_tools
+from pki_tools import types, exceptions
 
 
 def is_revoked(
     cert: Union[x509.Certificate, types.PemCert], crl_cache_seconds: int = 3600
 ) -> bool:
     if types._is_pem_str(cert):
-        cert = utils.cert_from_pem(cert)
+        cert = pki_tools.cert_from_pem(cert)
 
     ext = cert.extensions
     try:

pki_tools/ocsp.py

@@ -16,9 +16,8 @@
 from cryptography.x509.oid import ExtensionOID
 from loguru import logger
 
-from pki_tools import exceptions
-from pki_tools import utils
-from pki_tools import types
+import pki_tools
+from pki_tools import exceptions, types
 
 
 @lru_cache(maxsize=None)
@@ -32,18 +31,18 @@ def _get_issuer_from_uri(issuer_uri, cache_ttl=None):
             f"Issuer URI fetch failed. Status: {ret.status_code}"
         )
 
-    return utils.cert_from_pem(ret.text)
+    return pki_tools.cert_from_pem(ret.text)
 
 
 def is_revoked(
     cert: [x509.Certificate, types.PemCert],
     issuer_cert: [x509.Certificate, types.PemCert, types.OcspIssuerUri],
 ) -> bool:
     if types._is_pem_str(cert):
-        cert = utils.cert_from_pem(cert)
+        cert = pki_tools.cert_from_pem(cert)
 
     if types._is_pem_str(issuer_cert):
-        issuer_cert = utils.cert_from_pem(issuer_cert)
+        issuer_cert = pki_tools.cert_from_pem(issuer_cert)
     elif isinstance(issuer_cert, types.OcspIssuerUri):
         cache_ttl = round(time.time() / issuer_cert.cache_time_seconds)
         issuer_cert = _get_issuer_from_uri(

test/test_utils.py

@@ -5,7 +5,7 @@
 from cryptography.x509 import ocsp
 
 from pki_tools.exceptions import CertLoadError, Error
-from pki_tools.utils import cert_from_pem, is_revoked
+from pki_tools import cert_from_pem, is_revoked
 from conftest import _create_mocked_ocsp_response, _create_cert, _create_crl