Revize 1e07432d

Re #8476 - Implemented `list_certificates` with unit tests.

src/constants.py

@@ -2 +2 @@
 
 DATABASE_FILE = "db/database_sqlite.db"
 DATABASE_FILE_LOCATION = FileAnchor("aswi2021jmsd", DATABASE_FILE)
-
+DATETIME_FORMAT = "%m.%d.%Y %H:%M:%S"
 
 # Types of certificates
 ROOT_CA_ID = 1

swagger_server/controllers/certificates_controller.py

@@ +1 @@
+from datetime import datetime
+
 import connexion
 import six
 
@@ -8 +10 @@
 from src.services.cryptography import CryptographyService           # TODO not the Controller's responsibility. 2
 from sqlite3 import Connection                                      # TODO not the Controller's responsibility. 3
 from src.constants import DICT_USAGES, CA_ID, \
-    DATABASE_FILE_LOCATION                                          # TODO DATABASE_FILE - not the Controller's
+    DATABASE_FILE_LOCATION, SSL_ID, SIGNATURE_ID, AUTHENTICATION_ID, \
+    DATETIME_FORMAT  # TODO DATABASE_FILE - not the Controller's
                                                                     #  responsibility. 4
 from src.services.key_service import KeyService
-from swagger_server.models import CertificateRequest
+from swagger_server.models import CertificateRequest, CertificateListItem, CAUsage, IssuerListItem
 
 from swagger_server.models.certificate import Certificate  # noqa: E501
 from swagger_server.models.certificate_list_response import CertificateListResponse  # noqa: E501
@@ -31 +34 @@
                                                                         #  the one corresponding to the connection.
                                                                         #  The cursor can always be generated from the
                                                                         #  connection instance.
+GENERAL_ERROR = "An error occured during processing of the request."
+CORRUPTED_DATABASE = "Internal server error (corrupted database)."
 
 __ = CryptographyService()                                              # TODO not the Controller's responsibility. 6
 CERTIFICATE_SERVICE = CertificateService(__, CertificateRepository(None, None))
@@ -104 +109 @@
             if issuer_key is None:
                 return ErrorResponse(
                     success=False,
-                    data="Internal server error (corrupted database)."
+                    data=CORRUPTED_DATABASE
                 ), 400
 
             f = CERTIFICATE_SERVICE.create_ca if CA_ID in usages_dict and usages_dict[CA_ID] else \
@@ -130 +135 @@
                 data="The certificate could not have been created."
             ), 400
     else:
-        return 400
+        return ErrorResponse(success=False, data="The request must be JSON-formatted."), 400
 
 
 def get_certificate_by_id(id):  # noqa: E501
@@ -175 +180 @@
     """
     setup()                                                             # TODO remove after issue fixed
 
-    if connexion.request.is_json:
-        filtering = Filtering.from_dict(connexion.request.get_json())  # noqa: E501
-        print(filtering)
-        return 200
+    key_map = {CA_ID: 'ca', SSL_ID: 'ssl', SIGNATURE_ID: 'digital_signature', AUTHENTICATION_ID: 'authentication'}
+
+    if len(connexion.request.data) == 0:
+        certs = CERTIFICATE_SERVICE.get_certificates()
+        if certs is None:
+            return ErrorResponse(success=False, data=GENERAL_ERROR), 500
+        elif len(certs) == 0:
+            return ErrorResponse(success=False, data="No certificates found."), 204
+        else:
+            ret = []
+            for c in certs:
+                c_issuer = CERTIFICATE_SERVICE.get_certificate(c.parent_id)
+                if c_issuer is None:
+                    return ErrorResponse(success=False, data=CORRUPTED_DATABASE)
+
+                ret.append(
+                    CertificateListItem(
+                        id=c.certificate_id,
+                        cn=c.common_name,
+                        not_before=datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
+                        not_after=datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
+                        usage=CAUsage(**{key_map[k]: v for k, v in c.usages.items()}),
+                        issuer=IssuerListItem(
+                            id=c_issuer.certificate_id,
+                            cn=c_issuer.common_name
+                        )
+                    )
+                )
+            return CertificateListResponse(success=True, data=ret)
     else:
-        return 400
+        # TODO fix filtering issue (somehow)
+        return ErrorResponse(success=False, data="The request must be JSON-formatted."), 400
 
 
 def get_certificate_root_by_id(id):  # noqa: E501

tests/unit_tests/rest_api/certificates.py

@@ -123 +123 @@
     ret = server.post("/api/certificates", content_type="application/json", json={
         "subject": {
             "C": "CZ",
-            "CN": "Root CA s.r.o.",
+            "CN": "Another Root CA s.r.o.",
             "L": "Pilsen",
-            "O": "Root CA s.r.o.",
+            "O": "Another Root CA s.r.o.",
             "OU": "IT department",
             "ST": "Pilsen Region",
             "emailAddress": "[e-mail]"
@@ -149 +149 @@
     assert d["success"]
 
     ret = server.post("/api/certificates", content_type="application/json", json={
+        "CA": 2,
         "subject": {
-            "CA": 2,
             "C": "CZ",
-            "CN": "Root CA s.r.o.",
+            "CN": "Intermediate CA s.r.o.",
             "L": "Pilsen",
-            "O": "Root CA s.r.o.",
+            "O": "Intermediate CA s.r.o.",
             "OU": "IT department",
             "ST": "Pilsen Region",
-            "emailAddress": "[e-mail]"
+            "emailAddress": "[e-mail]"
         },
         "usage": {
             "CA": True,
@@ -175 +175 @@
     assert "data" in d
     assert d["data"] == 3
     assert "success" in d
+    assert d["success"]
+
+
+def test_list_of_certificates(server):
+    ret = server.get("/api/certificates")
+
+    assert ret.status_code == 200
+
+    usage_dict = {'CA': True, 'SSL': True, 'authentication': True, 'digitalSignature': True}
+    s1 = "Root CA s.r.o."
+    s2 = "Another Root CA s.r.o."
+    s3 = "Intermediate CA s.r.o."
+
+    assert "data" in ret.json
+    assert "success" in ret.json
+    assert ret.json["success"]
+
+    d = ret.json["data"]
+
+    # print(d)
+
+    assert d[0]['CN'] == s1
+    assert d[1]['CN'] == s2
+    assert d[2]['CN'] == s3
+
+    assert all(d[i]['id'] == i + 1 for i in range(3))
+    assert all(d[i]['usage'] == usage_dict for i in range(3))
+
+    assert d[0]['issuer']['CN'] == s1
+    assert d[1]['issuer']['CN'] == s2
+    assert d[2]['issuer']['CN'] == s2
+
+    assert d[0]['issuer']['id'] == 1
+    assert d[1]['issuer']['id'] == 2
+    assert d[2]['issuer']['id'] == 2
+
+
+def test_sign_by_non_ca(server):
+    ret = server.post("/api/certificates", content_type="application/json", json={
+        "CA": 2,
+        "subject": {
+            "C": "CZ",
+            "CN": "Fake Intermediate CA s.r.o.",
+            "L": "Pilsen",
+            "O": "Fake Intermediate CA s.r.o.",
+            "OU": "IT department",
+            "ST": "Pilsen Region",
+            "emailAddress": "[e-mail]"
+        },
+        "usage": {
+            "CA": False,
+            "SSL": True,
+            "authentication": True,
+            "digitalSignature": True
+        },
+        "validityDays": 30
+    })
+
+    assert ret.status_code == 201
+
+    d = ret.json
+
+    assert "data" in d
+    assert d["data"] == 4
+    assert "success" in d
+    assert d["success"]
+
+    ret = server.post("/api/certificates", content_type="application/json", json={
+        "CA": 2,
+        "subject": {
+            "C": "CZ",
+            "CN": "End certificate signed by end certificate s.r.o.",
+            "L": "Pilsen",
+            "O": "This is probably bad s.r.o.",
+            "OU": "IT department",
+            "ST": "Pilsen Region",
+            "emailAddress": "[e-mail]"
+        },
+        "usage": {
+            "CA": False,
+            "SSL": True,
+            "authentication": True,
+            "digitalSignature": True
+        },
+        "validityDays": 30
+    })
+
+    # TODO discussion -> assert ret.status_code == 400
+    assert ret.status_code == 201
+
+    d = ret.json
+
+    assert "data" in d
+    assert d["data"] == 5
+    assert "success" in d
     assert d["success"]