|
|
def load_certificate(certificate):
with suppress(ValueError, TypeError):
return x509.load_pem_x509_certificate(base64.b64decode(
certificate.with_context(bin_size=False).pem_certificate
))
return False
self.issuer_cert_id = False
cert_data = {
certificate: {
'loaded': loaded_certificate,
'issuer_cn': issuer_cn,
}
for certificate in self.filtered('pem_certificate')
if (loaded_certificate := load_certificate(certificate))
if (issuer_cn := self._get_common_name(loaded_certificate, issuer=True))
}
if cert_data:
valid_certs = self.filtered(lambda c: c in cert_data)
possible_parents = self.with_context(active_test=False).env['certificate.certificate'].search([
*self.env['certificate.certificate']._check_company_domain(valid_certs.mapped('company_id')),
('subject_common_name', 'in', list({d['issuer_cn'] for d in cert_data.values()})),
('pem_certificate', '!=', False),
])
for cert, data in cert_data.items():
candidates = possible_parents.filtered_domain([
*self.env['certificate.certificate']._check_company_domain(cert.company_id),
('subject_common_name', '=', data['issuer_cn']),
# Exclude the certificate itself so a self-signed cert (subject == issuer)
# is not matched as its own issuer
('id', '!=', cert.id),
# Prefer the candidate with the furthest expiration date (most recent renewal)
]).sorted(key=lambda p: p.date_end or fields.Datetime.now(), reverse=True)
# A candidate whose key cryptographically signed this certificate.
issuer = candidates.filtered(
lambda candidate: (x509_candidate := load_certificate(candidate))
and self._is_issued_by(data['loaded'], x509_candidate)
)[:1]
# No mathematical proof: fall back to a
# candidate whose SKI matches the expected issuer key identifier (AKI).
expected_ski = self._get_authority_key_identifier(data['loaded'])
if not issuer and expected_ski:
issuer = candidates.filtered(
lambda candidate: (x509_candidate := load_certificate(candidate))
and self._get_subject_key_identifier(x509_candidate) == expected_ski
)[:1]
cert.issuer_cert_id = issuer.id
|
|
