Revize 5b57121e

Re #8476 - Refactored the application; swagger is not used for code generation anymore, REST API is being implemented from scratch. Migration only, fixing tests.

app.py

@@ -1 +1 @@
 from flask import Flask
 import os
+from src.controllers.certificates_controller import CertController
 
 app = Flask(__name__)
 
@@ -9 +10 @@
     return 'Welcome to the X.509 management application homepage!'
 
 
[e-mail]('/api/certificates', methods=["POST"])
+def create_certificate():
+    return CertController.create_certificate()
+
+
[e-mail]('/api/certificates', methods=["GET"])
+def get_cert_list():
+    return CertController.get_certificate_list()
+
+
 if __name__ == '__main__':
     host = "0.0.0.0"
     port = 5000
@@ -20 +31 @@
     if "FLASK_PORT" in os.environ:
         host = os.environ["FLASK_PORT"]
 
-    app.run(host=host, port=port)
-
+    app.run(host=host, port=port)

requirements.txt

@@ -1 +1 @@
 Flask==1.1.2
-pytest~=6.2.3
-
-connexion~=2.7.0
-connexion[swagger-ui]
-six~=1.15.0
+pytest~=6.2.3

src/controllers/certificates_controller.py

@@ +1 @@
+from datetime import datetime
+from itertools import chain
+from flask import request
+from src.dao.private_key_repository import PrivateKeyRepository
+from src.model.subject import Subject
+from src.services.certificate_service import CertificateService
+from src.dao.certificate_repository import CertificateRepository  # TODO not the Controller's responsibility. 1
+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, SSL_ID, SIGNATURE_ID, AUTHENTICATION_ID, \
+    DATETIME_FORMAT, ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID  # TODO DATABASE_FILE - not the Controller's
+#  responsibility. 4
+from src.services.key_service import KeyService
+
+# _ = Connection("../" + DATABASE_FILE)                                 # TODO not the Controller's responsibility. 5
+# cursor = _.cursor()                                                   # TODO responsibility of the
+                                                                        #  CertificateRepository. It makes no sense to
+                                                                        #  supply a different cursor than precisely
+                                                                        #  the one corresponding to the connection.
+                                                                        #  The cursor can always be generated from the
+                                                                        #  connection instance.
+FILTERING = "filtering"
+ISSUER = "issuer"
+US = "usage"
+NOT_AFTER = "notAfter"
+NOT_BEFORE = "notBefore"
+COMMON_NAME = "CN"
+ID = "id"
+E_NO_ISSUER_FOUND = {"success": False,
+                           "data": "No certificate authority with such unique ID exists."}
+E_NO_CERTIFICATES_FOUND = {"success": False, "data": "No certificates found."}
+E_NOT_JSON_FORMAT = {"success": False, "data": "The request must be JSON-formatted."}
+E_CORRUPTED_DATABASE = {"success": False, "data": "Internal server error (corrupted database)."}
+E_GENERAL_ERROR = {"success": False, "data": "Internal server error (unknown origin)."}
+E_MISSING_PARAMETERS = {"success": False, "data": "Invalid request, missing parameters."}
+E_WRONG_PARAMETERS = {"success": False, "data": "Invalid request, wrong parameters."}
+USAGE = "usage"
+SUBJECT = "subject"
+VALIDITY_DAYS = "validityDays"
+CA = "CA"
+
+__ = CryptographyService()  # TODO not the Controller's responsibility. 6
+CERTIFICATE_SERVICE = CertificateService(__, CertificateRepository(None, None))
+# TODO open for discussion. Expected:
+#  CS = CertificateService.get_instance()
+#  or something like that.
+
+KEY_SERVICE = KeyService(__, PrivateKeyRepository(None, None))  # TODO as above
+
+
+class CertController:
+
+    @staticmethod
+    def setup():
+        """
+        SQLite3 thread issue hack.
+        :return:
+        """
+        _ = Connection(DATABASE_FILE_LOCATION.shortest_relative_path())
+        CERTIFICATE_SERVICE.certificate_repository.connection = _
+        CERTIFICATE_SERVICE.certificate_repository.cursor = _.cursor()
+        KEY_SERVICE.private_key_repository.connection = _
+        KEY_SERVICE.private_key_repository.cursor = _.cursor()
+
+    @staticmethod
+    def create_certificate():  # noqa: E501
+        """create new certificate
+
+        Create a new certificate based on given information # noqa: E501
+
+        :param body: Certificate data to be created
+        :type body: dict | bytes
+
+        :rtype: CreatedResponse
+        """
+        CertController.setup()  # TODO remove after issue fixed
+
+        key_map = {'CA': CA_ID, 'SSL': SSL_ID, 'digitalSignature': SIGNATURE_ID, 'authentication': AUTHENTICATION_ID}
+        required_keys = {SUBJECT, USAGE, VALIDITY_DAYS}
+
+        if request.is_json:
+            body = request.get_json()
+            if not all(k in body for k in required_keys):
+                return E_MISSING_PARAMETERS, 400
+
+            if not isinstance(body[VALIDITY_DAYS], int):
+                return E_WRONG_PARAMETERS, 400
+
+            subject = Subject.from_dict(body[SUBJECT])
+
+            if subject is None:
+                return E_WRONG_PARAMETERS, 400
+
+            usages_dict = DICT_USAGES.copy()
+
+            if not isinstance(body[USAGE], dict):
+                return E_WRONG_PARAMETERS, 400
+
+            for k, v in body[USAGE].items():
+                if k not in key_map:
+                    return E_WRONG_PARAMETERS, 400
+                usages_dict[key_map[k]] = v
+
+            key = KEY_SERVICE.create_new_key()  # TODO pass key
+
+            if CA not in body:
+                cert = CERTIFICATE_SERVICE.create_root_ca(
+                    key,
+                    subject,
+                    usages=usages_dict,
+                    days=body[VALIDITY_DAYS]
+                )
+            else:
+                issuer = CERTIFICATE_SERVICE.get_certificate(body[CA])
+
+                if issuer is None:
+                    return E_NO_ISSUER_FOUND, 400
+
+                issuer_key = KEY_SERVICE.get_key(issuer.private_key_id)
+
+                if issuer_key is None:
+                    return E_CORRUPTED_DATABASE, 500
+
+                f = CERTIFICATE_SERVICE.create_ca if CA_ID in usages_dict and usages_dict[CA_ID] else \
+                    CERTIFICATE_SERVICE.create_end_cert
+
+                cert = f(
+                    key,
+                    subject,
+                    issuer,
+                    issuer_key,
+                    usages=usages_dict,
+                    days=body[VALIDITY_DAYS]
+                )
+
+            if cert is not None:
+                return {"success": True,
+                        "data": cert.certificate_id}, 201
+            else:
+                return {"success": False,
+                        "data": "Internal error: The certificate could not have been created."}, 400
+        else:
+            return E_NOT_JSON_FORMAT, 400
+
+    @staticmethod
+    def get_certificate_by_id(id):  # noqa: E501
+        """get certificate by ID
+
+        Get certificate in PEM format by ID # noqa: E501
+
+        :param id: ID of a certificate to be queried
+        :type id: dict | bytes
+
+        :rtype: PemResponse
+        """
+        CertController.setup()  # TODO remove after issue fixed
+
+        if connexion.request.is_json:
+            id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
+        return 'do some magic!'
+
+    @staticmethod
+    def get_certificate_details_by_id(id):  # noqa: E501
+        """get certificate's details by ID
+
+        Get certificate details by ID # noqa: E501
+
+        :param id: ID of a certificate whose details are to be queried
+        :type id: dict | bytes
+
+        :rtype: CertificateResponse
+        """
+        CertController.setup()  # TODO remove after issue fixed
+
+        if connexion.request.is_json:
+            id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
+        return 'do some magic!'
+
+    @staticmethod
+    def get_certificate_list():  # noqa: E501
+        """get list of certificates
+
+        Lists certificates based on provided filtering options # noqa: E501
+
+        :param filtering: Filter certificate type to be queried
+        :type filtering: dict | bytes
+
+        :rtype: CertificateListResponse
+        """
+        CertController.setup()  # TODO remove after issue fixed
+
+        key_map = {CA_ID: 'CA', SSL_ID: 'SSL', SIGNATURE_ID: 'digitalSignature', AUTHENTICATION_ID: 'authentication'}
+
+        targets = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}
+        if request.is_json:
+            data = request.get_json()
+            if FILTERING in data:
+                if isinstance(data[FILTERING], dict):
+                    if CA in data[FILTERING]:
+                        if isinstance(data[FILTERING][CA], bool):
+                            if data[FILTERING][CA]:
+                                targets.remove(CERTIFICATE_ID)
+                            else:
+                                targets.remove(ROOT_CA_ID)
+                                targets.remove(INTERMEDIATE_CA_ID)
+                        else:
+                            return E_WRONG_PARAMETERS, 400
+                else:
+                    return E_WRONG_PARAMETERS, 400
+
+        if len(targets) == 3:
+            certs = CERTIFICATE_SERVICE.get_certificates()
+        else:
+            certs = list(chain(CERTIFICATE_SERVICE.get_certificates(target) for target in targets))
+
+        if certs is None:
+            return E_GENERAL_ERROR, 500
+        elif len(certs) == 0:
+            return E_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 E_CORRUPTED_DATABASE, 500
+
+                ret.append(
+                    {
+                        ID: c.certificate_id,
+                        COMMON_NAME: c.common_name,
+                        NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
+                        NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
+                        USAGE: {key_map[k]: v for k, v in c.usages.items()},
+                        ISSUER: {
+                            ID: c_issuer.certificate_id,
+                            COMMON_NAME: c_issuer.common_name
+                        }
+                    }
+                )
+            return {"success": True, "data": ret}
+
+
+
+    @staticmethod
+    def get_certificate_root_by_id(id):  # noqa: E501
+        """get certificate's root of trust chain by ID
+
+        Get certificate's root of trust chain in PEM format by ID # noqa: E501
+
+        :param id: ID of a child certificate whose root is to be queried
+        :type id: dict | bytes
+
+        :rtype: PemResponse
+        """
+        CertController.setup()  # TODO remove after issue fixed
+
+        if connexion.request.is_json:
+            id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
+        return 'do some magic!'
+
+    @staticmethod
+    def get_certificate_trust_chain_by_id(id):  # noqa: E501
+        """get certificate's trust chain by ID
+
+        Get certificate trust chain in PEM format by ID # noqa: E501
+
+        :param id: ID of a child certificate whose chain is to be queried
+        :type id: dict | bytes
+
+        :rtype: PemResponse
+        """
+        CertController.setup()  # TODO remove after issue fixed
+
+        if connexion.request.is_json:
+            id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
+        return 'do some magic!'

src/model/subject.py

@@ -1 +1 @@
 class Subject:
+    ATTR_MAP = {"C": "common_name", "ST": "state", "L": "locality", "CN": "common_name", "O": "organization",
+                "OU": "organization_unit", "emailAddress": "email_address"}
 
     def __init__(self, common_name=None, country=None, locality=None, state=None, organization=None,
                  organization_unit=None, email_address=None):
@@ -9 +11 @@
         self.organization = organization
         self.organization_unit = organization_unit
         self.email_address = email_address
+
+    @staticmethod
+    def from_dict(d):
+        if not isinstance(d, dict):
+            return None
+
+        s = Subject()
+        for k, v in Subject.ATTR_MAP.items():
+            if k in d:
+                if not isinstance(d[k], str):
+                    return None
+                s.__setattr__(v, d[k])
+
+        return s

src/swagger.yaml

@@ +1 @@
+openapi: 3.0.0
+info:
+  title: X.509 certificate management
+  description: API for certificate management created for YOSO company
+  version: 1.0.1
+servers:
+- url: https://virtserver.swaggerhub.com/janpasek97/X509_management/1.0.1
+  description: X509 management API
+tags:
+- name: certificates
+  description: API for creating and querying certificates
+paths:
+  /api/certificates/{id}:
+    get:
+      tags:
+      - certificates
+      summary: get certificate by ID
+      description: Get certificate in PEM format by ID
+      operationId: get_certificate_by_id
+      parameters:
+      - name: id
+        in: path
+        description: ID of a certificate to be queried
+        required: true
+        style: simple
+        explode: false
+        schema:
+          $ref: '#/components/schemas/IdParameter'
+      responses:
+        "200":
+          description: returning the certificate
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/PemResponse'
+        "204":
+          description: the certificate was not found
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+        "400":
+          description: bad request
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+      x-openapi-router-controller: swagger_server.controllers.certificates_controller
+  /api/certificates/{id}/chain:
+    get:
+      tags:
+      - certificates
+      summary: get certificate's trust chain by ID
+      description: Get certificate trust chain in PEM format by ID
+      operationId: get_certificate_trust_chain_by_id
+      parameters:
+      - name: id
+        in: path
+        description: ID of a child certificate whose chain is to be queried
+        required: true
+        style: simple
+        explode: false
+        schema:
+          $ref: '#/components/schemas/IdParameter'
+      responses:
+        "200":
+          description: returning the trust chain
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/PemResponse'
+        "204":
+          description: the certificate was not found
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+        "400":
+          description: bad request
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+      x-openapi-router-controller: swagger_server.controllers.certificates_controller
+  /api/certificates/{id}/root:
+    get:
+      tags:
+      - certificates
+      summary: get certificate's root of trust chain by ID
+      description: Get certificate's root of trust chain in PEM format by ID
+      operationId: get_certificate_root_by_id
+      parameters:
+      - name: id
+        in: path
+        description: ID of a child certificate whose root is to be queried
+        required: true
+        style: simple
+        explode: false
+        schema:
+          $ref: '#/components/schemas/IdParameter'
+      responses:
+        "200":
+          description: returning the root of trust chain
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/PemResponse'
+        "204":
+          description: the certificate was not found
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+        "400":
+          description: bad request
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+      x-openapi-router-controller: swagger_server.controllers.certificates_controller
+  /api/certificates/{id}/details:
+    get:
+      tags:
+      - certificates
+      summary: get certificate's details by ID
+      description: Get certificate details by ID
+      operationId: get_certificate_details_by_id
+      parameters:
+      - name: id
+        in: path
+        description: ID of a certificate whose details are to be queried
+        required: true
+        style: simple
+        explode: false
+        schema:
+          $ref: '#/components/schemas/IdParameter'
+      responses:
+        "200":
+          description: returning the certificate details
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/CertificateResponse'
+        "204":
+          description: the certificate was not found
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+        "400":
+          description: bad request
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+      x-openapi-router-controller: swagger_server.controllers.certificates_controller
+  /api/certificates:
+    get:
+      tags:
+      - certificates
+      summary: get list of certificates
+      description: Lists certificates based on provided filtering options
+      operationId: get_certificate_list
+      parameters:
+      - name: filtering
+        in: query
+        description: Filter certificate type to be queried
+        required: false
+        style: form
+        explode: true
+        schema:
+          $ref: '#/components/schemas/Filtering'
+      responses:
+        "200":
+          description: returning results matching filtering criteria
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/CertificateListResponse'
+        "204":
+          description: no certificates found
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+        "400":
+          description: bad request
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+      x-openapi-router-controller: swagger_server.controllers.certificates_controller
+    post:
+      tags:
+      - certificates
+      summary: create new certificate
+      description: Create a new certificate based on given information
+      operationId: create_certificate
+      requestBody:
+        description: Certificate data to be created
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/CertificateRequest'
+      responses:
+        "201":
+          description: item created
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/CreatedResponse'
+        "400":
+          description: "invalid input, object invalid"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+        "409":
+          description: an existing item already exists
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+      x-openapi-router-controller: swagger_server.controllers.certificates_controller
+components:
+  schemas:
+    CAUsage:
+      required:
+      - CA
+      - SSL
+      - authentication
+      - digitalSignature
+      properties:
+        CA:
+          type: boolean
+        authentication:
+          type: boolean
+        digitalSignature:
+          type: boolean
+        SSL:
+          type: boolean
+      example:
+        digitalSignature: true
+        SSL: true
+        CA: true
+        authentication: true
+    IssuerListItem:
+      required:
+      - CN
+      - id
+      properties:
+        id:
+          type: integer
+          example: 547
+        CN:
+          type: string
+          example: Root CA s.r.o.
+      example:
+        id: 547
+        CN: Root CA s.r.o.
+    CertificateListResponse:
+      properties:
+        success:
+          type: boolean
+          example: true
+        data:
+          type: array
+          items:
+            $ref: '#/components/schemas/CertificateListItem'
+      example:
+        data:
+        - notAfter: 2021-07-01T00:00:00.000+00:00
+          usage:
+            digitalSignature: true
+            SSL: true
+            CA: true
+            authentication: true
+          id: 547
+          CN: Root CA s.r.o.
+          notBefore: 2021-03-31T00:00:00.000+00:00
+          issuer:
+            id: 547
+            CN: Root CA s.r.o.
+        - notAfter: 2021-07-01T00:00:00.000+00:00
+          usage:
+            digitalSignature: true
+            SSL: true
+            CA: true
+            authentication: true
+          id: 547
+          CN: Root CA s.r.o.
+          notBefore: 2021-03-31T00:00:00.000+00:00
+          issuer:
+            id: 547
+            CN: Root CA s.r.o.
+        success: true
+    CertificateListItem:
+      properties:
+        id:
+          type: integer
+          example: 547
+        CN:
+          type: string
+          example: Root CA s.r.o.
+        notBefore:
+          type: string
+          format: date
+          example: 2021-03-31
+        notAfter:
+          type: string
+          format: date
+          example: 2021-07-01
+        usage:
+          $ref: '#/components/schemas/CAUsage'
+        issuer:
+          $ref: '#/components/schemas/IssuerListItem'
+      example:
+        notAfter: 2021-07-01T00:00:00.000+00:00
+        usage:
+          digitalSignature: true
+          SSL: true
+          CA: true
+          authentication: true
+        id: 547
+        CN: Root CA s.r.o.
+        notBefore: 2021-03-31T00:00:00.000+00:00
+        issuer:
+          id: 547
+          CN: Root CA s.r.o.
+    Filtering:
+      properties:
+        CA:
+          type: boolean
+    Subject:
+      required:
+      - CN
+      properties:
+        C:
+          type: string
+          description: Country code
+          example: CZ
+        ST:
+          type: string
+          description: State/Province
+          example: Pilsen Region
+        L:
+          type: string
+          description: Locality
+          example: Pilsen
+        CN:
+          type: string
+          description: Common name
+          example: Root CA s.r.o.
+        O:
+          type: string
+          description: Organization
+          example: Root CA s.r.o.
+        OU:
+          type: string
+          description: Organization Unit
+          example: IT department
+        emailAddress:
+          type: string
+          description: Email Address
+          example: [e-mail]
+      example:
+        ST: Pilsen Region
+        emailAddress: [e-mail]
+        C: CZ
+        OU: IT department
+        CN: Root CA s.r.o.
+        L: Pilsen
+        O: Root CA s.r.o.
+    Certificate:
+      required:
+      - notAfter
+      - notBefore
+      - subject
+      - usage
+      properties:
+        subject:
+          $ref: '#/components/schemas/Subject'
+        notBefore:
+          type: string
+          format: date
+          example: 2021-03-31
+        notAfter:
+          type: string
+          format: date
+          example: 2021-07-01
+        usage:
+          $ref: '#/components/schemas/CAUsage'
+        CA:
+          type: integer
+          description: ID of the new item
+          example: 547
+      example:
+        notAfter: 2021-07-01T00:00:00.000+00:00
+        subject:
+          ST: Pilsen Region
+          emailAddress: [e-mail]
+          C: CZ
+          OU: IT department
+          CN: Root CA s.r.o.
+          L: Pilsen
+          O: Root CA s.r.o.
+        usage:
+          digitalSignature: true
+          SSL: true
+          CA: true
+          authentication: true
+        notBefore: 2021-03-31T00:00:00.000+00:00
+        CA: 547
+    CertificateRequest:
+      required:
+      - subject
+      - usage
+      - validityDays
+      properties:
+        subject:
+          $ref: '#/components/schemas/Subject'
+        validityDays:
+          type: integer
+          example: 30
+        usage:
+          $ref: '#/components/schemas/CAUsage'
+        CA:
+          type: integer
+          description: ID of the new item
+          example: 547
+    CreatedResponse:
+      required:
+      - data
+      - success
+      properties:
+        success:
+          type: boolean
+          example: true
+        data:
+          type: integer
+          example: 457
+      description: Item was created
+      example:
+        data: 457
+        success: true
+    ErrorResponse:
+      required:
+      - data
+      - success
+      properties:
+        success:
+          type: boolean
+          example: false
+        data:
+          type: string
+          example: An error occured
+    CertificateResponse:
+      required:
+      - data
+      - success
+      properties:
+        success:
+          type: boolean
+          example: true
+        data:
+          $ref: '#/components/schemas/Certificate'
+      example:
+        data:
+          notAfter: 2021-07-01T00:00:00.000+00:00
+          subject:
+            ST: Pilsen Region
+            emailAddress: [e-mail]
+            C: CZ
+            OU: IT department
+            CN: Root CA s.r.o.
+            L: Pilsen
+            O: Root CA s.r.o.
+          usage:
+            digitalSignature: true
+            SSL: true
+            CA: true
+            authentication: true
+          notBefore: 2021-03-31T00:00:00.000+00:00
+          CA: 547
+        success: true
+    PemResponse:
+      required:
+      - data
+      - success
+      properties:
+        success:
+          type: boolean
+          example: true
+        data:
+          type: string
+          description: Single PEM file or concatenation of multiple PEM formatted
+            certificates
+          example: '-----BEGIN CERTIFICATE-----MIICLDCCAdKgAwIBAgIBADAKBggqhkjOPQQDAjB9MQswCQYDVQQGEwJCRTEPMA0GA1UEChMGR251VExTMSUwIwYDVQQ...etc-----END
+            CERTIFICATE-----'
+      example:
+        data: '-----BEGIN CERTIFICATE-----MIICLDCCAdKgAwIBAgIBADAKBggqhkjOPQQDAjB9MQswCQYDVQQGEwJCRTEPMA0GA1UEChMGR251VExTMSUwIwYDVQQ...etc-----END
+          CERTIFICATE-----'
+        success: true
+    IdParameter:
+      required:
+      - id
+      properties:
+        id:
+          type: integer
+          example: 444
+

src/utils/util.py

@@ +1 @@
+import datetime
+
+import six
+import typing
+
+
+def _deserialize(data, klass):
+    """Deserializes dict, list, str into an object.
+
+    :param data: dict, list or str.
+    :param klass: class literal, or string of class name.
+
+    :return: object.
+    """
+    if data is None:
+        return None
+
+    if klass in six.integer_types or klass in (float, str, bool):
+        return _deserialize_primitive(data, klass)
+    elif klass == object:
+        return _deserialize_object(data)
+    elif klass == datetime.date:
+        return deserialize_date(data)
+    elif klass == datetime.datetime:
+        return deserialize_datetime(data)
+    elif hasattr(klass, '__origin__'):
+        if klass.__origin__ == list:
+            return _deserialize_list(data, klass.__args__[0])
+        if klass.__origin__ == dict:
+            return _deserialize_dict(data, klass.__args__[1])
+    else:
+        return deserialize_model(data, klass)
+
+
+def _deserialize_primitive(data, klass):
+    """Deserializes to primitive type.
+
+    :param data: data to deserialize.
+    :param klass: class literal.
+
+    :return: int, long, float, str, bool.
+    :rtype: int | long | float | str | bool
+    """
+    try:
+        value = klass(data)
+    except UnicodeEncodeError:
+        value = six.u(data)
+    except TypeError:
+        value = data
+    return value
+
+
+def _deserialize_object(value):
+    """Return a original value.
+
+    :return: object.
+    """
+    return value
+
+
+def deserialize_date(string):
+    """Deserializes string to date.
+
+    :param string: str.
+    :type string: str
+    :return: date.
+    :rtype: date
+    """
+    try:
+        from dateutil.parser import parse
+        return parse(string).date()
+    except ImportError:
+        return string
+
+
+def deserialize_datetime(string):
+    """Deserializes string to datetime.
+
+    The string should be in iso8601 datetime format.
+
+    :param string: str.
+    :type string: str
+    :return: datetime.
+    :rtype: datetime
+    """
+    try:
+        from dateutil.parser import parse
+        return parse(string)
+    except ImportError:
+        return string
+
+
+def deserialize_model(data, klass):
+    """Deserializes list or dict to model.
+
+    :param data: dict, list.
+    :type data: dict | list
+    :param klass: class literal.
+    :return: model object.
+    """
+    instance = klass()
+
+    if not instance.swagger_types:
+        return data
+
+    for attr, attr_type in six.iteritems(instance.swagger_types):
+        if data is not None \
+                and instance.attribute_map[attr] in data \
+                and isinstance(data, (list, dict)):
+            value = data[instance.attribute_map[attr]]
+            setattr(instance, attr, _deserialize(value, attr_type))
+
+    return instance
+
+
+def _deserialize_list(data, boxed_type):
+    """Deserializes a list and its elements.
+
+    :param data: list to deserialize.
+    :type data: list
+    :param boxed_type: class literal.
+
+    :return: deserialized list.
+    :rtype: list
+    """
+    return [_deserialize(sub_data, boxed_type)
+            for sub_data in data]
+
+
+def _deserialize_dict(data, boxed_type):
+    """Deserializes a dict and its elements.
+
+    :param data: dict to deserialize.
+    :type data: dict
+    :param boxed_type: class literal.
+
+    :return: deserialized dict.
+    :rtype: dict
+    """
+    return {k: _deserialize(v, boxed_type)
+            for k, v in six.iteritems(data)}

swagger_server/__main__.py

@@ -1 @@
-#!/usr/bin/env python3
-import logging
-
-import connexion
-
-from swagger_server import encoder
-
-
-app = connexion.App(__name__, specification_dir='./swagger/', debug=False)
-PORT = 8080
-
-
-def setup(a):
-    a.app.json_encoder = encoder.JSONEncoder
-    a.add_api('swagger.yaml', arguments={'title': 'X.509 certificate management'}, base_path="/", pythonic_params=True)
-    # a.run(port=PORT)
-
-
-if __name__ == '__main__':
-    setup(app)
-    app.run(PORT)

swagger_server/controllers/authorization_controller.py

@@ -1 @@
-from typing import List
-"""
-controller generated to handled auth operation described at:
-https://connexion.readthedocs.io/en/latest/security.html
-"""
-

swagger_server/controllers/certificates_controller.py

@@ -1 @@
-from datetime import datetime
-
-import connexion
-import six
-
-from src.dao.private_key_repository import PrivateKeyRepository
-from src.model.subject import Subject
-from src.services.certificate_service import CertificateService
-from src.dao.certificate_repository import CertificateRepository    # TODO not the Controller's responsibility. 1
-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, 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, CertificateListItem, CAUsage, IssuerListItem
-
-from swagger_server.models.certificate import Certificate  # noqa: E501
-from swagger_server.models.certificate_list_response import CertificateListResponse  # noqa: E501
-from swagger_server.models.certificate_response import CertificateResponse  # noqa: E501
-from swagger_server.models.created_response import CreatedResponse  # noqa: E501
-from swagger_server.models.error_response import ErrorResponse  # noqa: E501
-from swagger_server.models.filtering import Filtering  # noqa: E501
-from swagger_server.models.id_parameter import IdParameter  # noqa: E501
-from swagger_server.models.pem_response import PemResponse  # noqa: E501
-from swagger_server import util
-
-
-# _ = Connection("../" + DATABASE_FILE)                                 # TODO not the Controller's responsibility. 5
-# cursor = _.cursor()                                                   # TODO responsibility of the
-                                                                        #  CertificateRepository. It makes no sense to
-                                                                        #  supply a different cursor than precisely
-                                                                        #  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))
-                                                                        # TODO open for discussion. Expected:
-                                                                        #  CS = CertificateService.get_instance()
-                                                                        #  or something like that.
-
-KEY_SERVICE = KeyService(__, PrivateKeyRepository(None, None))          # TODO as above
-
-
-
-def setup():
-    """
-    SQLite3 thread issue hack.
-    :return:
-    """
-    _ = Connection(DATABASE_FILE_LOCATION.shortest_relative_path())
-    CERTIFICATE_SERVICE.certificate_repository.connection = _
-    CERTIFICATE_SERVICE.certificate_repository.cursor = _.cursor()
-    KEY_SERVICE.private_key_repository.connection = _
-    KEY_SERVICE.private_key_repository.cursor = _.cursor()
-
-
-def create_certificate(body=None):  # noqa: E501
-    """create new certificate
-
-    Create a new certificate based on given information # noqa: E501
-
-    :param body: Certificate data to be created
-    :type body: dict | bytes
-
-    :rtype: CreatedResponse
-    """
-    setup()                                                             # TODO remove after issue fixed
-
-    if connexion.request.is_json:
-        body = CertificateRequest.from_dict(connexion.request.get_json())  # noqa: E501
-        # if body.subject is None or body.usage is None or body.validity_days is None:
-        #     return 400
-
-        key = KEY_SERVICE.create_new_key()                              # TODO pass key
-        subject = Subject(
-            common_name=body.subject.cn,
-            country=body.subject.c,
-            locality=body.subject.l,
-            state=body.subject.st,
-            organization=body.subject.o,
-            organization_unit=body.subject.ou,
-            email_address=body.subject.email_address
-        )
-        usages_dict = DICT_USAGES.copy()
-
-        if body.ca is None:
-            cert = CERTIFICATE_SERVICE.create_root_ca(
-                key,
-                subject,
-                usages=usages_dict,
-                days=body.validity_days
-            )
-        else:
-            issuer = CERTIFICATE_SERVICE.get_certificate(body.ca)
-
-            if issuer is None:
-                return ErrorResponse(
-                    success=False,
-                    data="No certificate authority with such unique ID exists."
-                ), 400
-
-            issuer_key = KEY_SERVICE.get_key(issuer.private_key_id)
-
-            if issuer_key is None:
-                return ErrorResponse(
-                    success=False,
-                    data=CORRUPTED_DATABASE
-                ), 400
-
-            f = CERTIFICATE_SERVICE.create_ca if CA_ID in usages_dict and usages_dict[CA_ID] else \
-                CERTIFICATE_SERVICE.create_end_cert
-
-            cert = f(
-                key,
-                subject,
-                issuer,
-                issuer_key,
-                usages=usages_dict,
-                days=body.validity_days
-            )
-
-        if cert is not None:
-            return CreatedResponse(
-                success=True,
-                data=cert.certificate_id
-            ), 201
-        else:
-            return ErrorResponse(
-                success=False,
-                data="The certificate could not have been created."
-            ), 400
-    else:
-        return ErrorResponse(success=False, data="The request must be JSON-formatted."), 400
-
-
-def get_certificate_by_id(id):  # noqa: E501
-    """get certificate by ID
-
-    Get certificate in PEM format by ID # noqa: E501
-
-    :param id: ID of a certificate to be queried
-    :type id: dict | bytes
-
-    :rtype: PemResponse
-    """
-    if connexion.request.is_json:
-        id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
-    return 'do some magic!'
-
-
-def get_certificate_details_by_id(id):  # noqa: E501
-    """get certificate's details by ID
-
-    Get certificate details by ID # noqa: E501
-
-    :param id: ID of a certificate whose details are to be queried
-    :type id: dict | bytes
-
-    :rtype: CertificateResponse
-    """
-    if connexion.request.is_json:
-        id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
-    return 'do some magic!'
-
-
-def get_certificate_list(filtering=None):  # noqa: E501
-    """get list of certificates
-
-    Lists certificates based on provided filtering options # noqa: E501
-
-    :param filtering: Filter certificate type to be queried
-    :type filtering: dict | bytes
-
-    :rtype: CertificateListResponse
-    """
-    setup()                                                             # TODO remove after issue fixed
-
-    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:
-        # 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
-    """get certificate's root of trust chain by ID
-
-    Get certificate's root of trust chain in PEM format by ID # noqa: E501
-
-    :param id: ID of a child certificate whose root is to be queried
-    :type id: dict | bytes
-
-    :rtype: PemResponse
-    """
-    if connexion.request.is_json:
-        id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
-    return 'do some magic!'
-
-
-def get_certificate_trust_chain_by_id(id):  # noqa: E501
-    """get certificate's trust chain by ID
-
-    Get certificate trust chain in PEM format by ID # noqa: E501
-
-    :param id: ID of a child certificate whose chain is to be queried
-    :type id: dict | bytes
-
-    :rtype: PemResponse
-    """
-    if connexion.request.is_json:
-        id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
-    return 'do some magic!'

swagger_server/encoder.py

@@ -1 @@
-from connexion.apps.flask_app import FlaskJSONEncoder
-import six
-
-from swagger_server.models.base_model_ import Model
-
-
-class JSONEncoder(FlaskJSONEncoder):
-    include_nulls = False
-
-    def default(self, o):
-        if isinstance(o, Model):
-            dikt = {}
-            for attr, _ in six.iteritems(o.swagger_types):
-                value = getattr(o, attr)
-                if value is None and not self.include_nulls:
-                    continue
-                attr = o.attribute_map[attr]
-                dikt[attr] = value
-            return dikt
-        return FlaskJSONEncoder.default(self, o)

swagger_server/models/__init__.py

@@ -1 @@
-# coding: utf-8
-
-# flake8: noqa
-from __future__ import absolute_import
-# import models into model package
-from swagger_server.models.ca_usage import CAUsage
-from swagger_server.models.certificate import Certificate
-from swagger_server.models.certificate_list_item import CertificateListItem
-from swagger_server.models.certificate_list_response import CertificateListResponse
-from swagger_server.models.certificate_request import CertificateRequest
-from swagger_server.models.certificate_response import CertificateResponse
-from swagger_server.models.created_response import CreatedResponse
-from swagger_server.models.error_response import ErrorResponse
-from swagger_server.models.filtering import Filtering
-from swagger_server.models.id_parameter import IdParameter
-from swagger_server.models.issuer_list_item import IssuerListItem
-from swagger_server.models.pem_response import PemResponse
-from swagger_server.models.subject import Subject

swagger_server/models/base_model_.py

@@ -1 @@
-import pprint
-
-import six
-import typing
-
-from swagger_server import util
-
-T = typing.TypeVar('T')
-
-
-class Model(object):
-    # swaggerTypes: The key is attribute name and the
-    # value is attribute type.
-    swagger_types = {}
-
-    # attributeMap: The key is attribute name and the
-    # value is json key in definition.
-    attribute_map = {}
-
-    @classmethod
-    def from_dict(cls: typing.Type[T], dikt) -> T:
-        """Returns the dict as a model"""
-        return util.deserialize_model(dikt, cls)
-
-    def to_dict(self):
-        """Returns the model properties as a dict
-
-        :rtype: dict
-        """
-        result = {}
-
-        for attr, _ in six.iteritems(self.swagger_types):
-            value = getattr(self, attr)
-            if isinstance(value, list):
-                result[attr] = list(map(
-                    lambda x: x.to_dict() if hasattr(x, "to_dict") else x,
-                    value
-                ))
-            elif hasattr(value, "to_dict"):
-                result[attr] = value.to_dict()
-            elif isinstance(value, dict):
-                result[attr] = dict(map(
-                    lambda item: (item[0], item[1].to_dict())
-                    if hasattr(item[1], "to_dict") else item,
-                    value.items()
-                ))
-            else:
-                result[attr] = value
-
-        return result
-
-    def to_str(self):
-        """Returns the string representation of the model
-
-        :rtype: str
-        """
-        return pprint.pformat(self.to_dict())
-
-    def __repr__(self):
-        """For `print` and `pprint`"""
-        return self.to_str()
-
-    def __eq__(self, other):
-        """Returns true if both objects are equal"""
-        return self.__dict__ == other.__dict__
-
-    def __ne__(self, other):
-        """Returns true if both objects are not equal"""
-        return not self == other

swagger_server/models/ca_usage.py

@@ -1 @@
-# coding: utf-8
-
-from __future__ import absolute_import
-from datetime import date, datetime  # noqa: F401
-
-from typing import List, Dict  # noqa: F401
-
-from swagger_server.models.base_model_ import Model
-from swagger_server import util
-
-
-class CAUsage(Model):
-    """NOTE: This class is auto generated by the swagger code generator program.
-
-    Do not edit the class manually.
-    """
-    def __init__(self, ca: bool=None, authentication: bool=None, digital_signature: bool=None, ssl: bool=None):  # noqa: E501
-        """CAUsage - a model defined in Swagger
-
-        :param ca: The ca of this CAUsage.  # noqa: E501
-        :type ca: bool
-        :param authentication: The authentication of this CAUsage.  # noqa: E501
-        :type authentication: bool
-        :param digital_signature: The digital_signature of this CAUsage.  # noqa: E501
-        :type digital_signature: bool
-        :param ssl: The ssl of this CAUsage.  # noqa: E501
-        :type ssl: bool
-        """
-        self.swagger_types = {
-            'ca': bool,
-            'authentication': bool,
-            'digital_signature': bool,
-            'ssl': bool
-        }
-
-        self.attribute_map = {
-            'ca': 'CA',
-            'authentication': 'authentication',
-            'digital_signature': 'digitalSignature',
-            'ssl': 'SSL'