|
1
|
from src.constants import ROOT_CA_ID, INTERMEDIATE_CA_ID
|
|
2
|
from src.dao.certificate_repository import CertificateRepository
|
|
3
|
from src.model.certificate import Certificate
|
|
4
|
from src.model.private_key import PrivateKey
|
|
5
|
from src.model.subject import Subject
|
|
6
|
from src.services.cryptography import CryptographyService
|
|
7
|
|
|
8
|
import time
|
|
9
|
|
|
10
|
DATE_FORMAT = "%d.%m.%Y %H:%M:%S"
|
|
11
|
CA_EXTENSIONS = "basicConstraints=critical,CA:TRUE"
|
|
12
|
|
|
13
|
|
|
14
|
class CertificateService:
|
|
15
|
|
|
16
|
def __init__(self, cryptography_service: CryptographyService, certificate_repository: CertificateRepository):
|
|
17
|
self.cryptography_service = cryptography_service
|
|
18
|
self.certificate_repository = certificate_repository
|
|
19
|
|
|
20
|
# TODO usages present in method parameters but not in class diagram
|
|
21
|
def create_root_ca(self, key: PrivateKey, subject: Subject, extensions: str = "", config: str = ""):
|
|
22
|
# create a new self signed certificate
|
|
23
|
cert_pem = self.cryptography_service.create_sscrt(subject, key.private_key, key_pass=key.password,
|
|
24
|
extensions=extensions, config=config)
|
|
25
|
|
|
26
|
# parse the generated pem for subject and notBefore/notAfter fields
|
|
27
|
subj, not_before, not_after = self.cryptography_service.parse_cert_pem(cert_pem)
|
|
28
|
|
|
29
|
# format the parsed date
|
|
30
|
not_before_formatted = time.strftime(DATE_FORMAT, not_before)
|
|
31
|
not_after_formatted = time.strftime(DATE_FORMAT, not_after)
|
|
32
|
|
|
33
|
# create a certificate wrapper
|
|
34
|
certificate = Certificate(-1, subject.common_name, not_before_formatted, not_after_formatted, cert_pem,
|
|
35
|
key.private_key_id, ROOT_CA_ID, 0, {})
|
|
36
|
|
|
37
|
# store the wrapper into the repository
|
|
38
|
created_id = self.certificate_repository.create(certificate)
|
|
39
|
|
|
40
|
# assign the generated ID to the inserted certificate
|
|
41
|
certificate.certificate_id = created_id
|
|
42
|
|
|
43
|
return certificate
|
|
44
|
|
|
45
|
# TODO config parameter present in class diagram but not here (unused)
|
|
46
|
def create_ca(self, subject_key: PrivateKey, subject: Subject, issuer_cert: Certificate, issuer_key: PrivateKey,
|
|
47
|
extensions: str = "", days: int = 30):
|
|
48
|
extensions = extensions + "\n" + CA_EXTENSIONS
|
|
49
|
# TODO implement AIA URI via extensions
|
|
50
|
cert_pem = self.cryptography_service.create_crt(subject, subject_key.private_key, issuer_cert.pem_data,
|
|
51
|
issuer_key.private_key,
|
|
52
|
subject_key_pass=subject_key.password,
|
|
53
|
issuer_key_pass=issuer_key.password, extensions=extensions,
|
|
54
|
days=days)
|
|
55
|
|
|
56
|
# parse the generated pem for subject and notBefore/notAfter fields
|
|
57
|
subj, not_before, not_after = self.cryptography_service.parse_cert_pem(cert_pem)
|
|
58
|
|
|
59
|
# format the parsed date
|
|
60
|
not_before_formatted = time.strftime(DATE_FORMAT, not_before)
|
|
61
|
not_after_formatted = time.strftime(DATE_FORMAT, not_after)
|
|
62
|
|
|
63
|
# create a certificate wrapper
|
|
64
|
certificate = Certificate(-1, subject.common_name, not_before_formatted, not_after_formatted, cert_pem,
|
|
65
|
subject_key.private_key_id, INTERMEDIATE_CA_ID, 0, {})
|
|
66
|
|
|
67
|
# store the wrapper into the repository
|
|
68
|
created_id = self.certificate_repository.create(certificate)
|
|
69
|
|
|
70
|
# assign the generated ID to the inserted certificate
|
|
71
|
certificate.certificate_id = created_id
|
|
72
|
|
|
73
|
return certificate
|
|
74
|
|
|
75
|
def get_certificate(self, unique_id: int) -> Certificate:
|
|
76
|
return self.certificate_repository.read(unique_id)
|