asyncua.crypto.certificate_handler.CertificateHandler.trust_certificate() - Code Metrics - Inspection of "User management with certificates" - FreeOpcUa/opcua-asyncio - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#209)

unknown

created 2020-06-09 13:25 UTC

CertificateHandler.trust_certificate() A

↳ Parent: asyncua.crypto.certificate_handler

Complexity

Conditions

Size

Total Lines	10
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	9
nop	5
dl	0
loc	10
rs	9.95
c	0
b	0
f	0

from asyncua.crypto import uacrypto
import sys
import logging
from asyncua.server.users import UserRole, User
sys.path.append('..')


class CertificateHandler:
    def __init__(self):
        self._trusted_certificates = {}

    async def trust_certificate(self, certificate_path: str, format: str = None, label: str = None,
                                user_role=UserRole.User):
        certificate = await uacrypto.load_certificate(certificate_path, format)
        if label is None:
            label = certificate_path
        user = User(role=user_role, name=label)
        if label in self._trusted_certificates:
            logging.warning(f"certificate with label {label} "
                            f"attempted to be added multiple times, only the last version will be kept.")
        self._trusted_certificates[label] = {'certificate': uacrypto.der_from_x509(certificate), 'user':user}

    def __contains__(self, certificate):
        return any(certificate == prospective_cert['certificate']
                   for prospective_cert
                   in self._trusted_certificates.values())

    def check_certificate(self, certificate):
        return certificate in self

    def get_user(self, certificate):
        correct_users = [prospective_certificate['user'] for prospective_certificate in self._trusted_certificates.values()
                         if certificate == prospective_certificate['certificate']]
        if len(correct_users) == 0:
            return None
        else:
            return correct_users[0]


1			from asyncua.crypto import uacrypto
2			import sys
3			import logging
4			from asyncua.server.users import UserRole, User
5			sys.path.append('..')
6
7
8			class CertificateHandler:
9			def __init__(self):
10			self._trusted_certificates = {}
11
12			async def trust_certificate(self, certificate_path: str, format: str = None, label: str = None,
13			user_role=UserRole.User):
14			certificate = await uacrypto.load_certificate(certificate_path, format)
15			if label is None:
16			label = certificate_path
17			user = User(role=user_role, name=label)
18			if label in self._trusted_certificates:
19			logging.warning(f"certificate with label {label} "
20			f"attempted to be added multiple times, only the last version will be kept.")
21			self._trusted_certificates[label] = {'certificate': uacrypto.der_from_x509(certificate), 'user':user}
22
23			def __contains__(self, certificate):
24			return any(certificate == prospective_cert['certificate']
25			for prospective_cert
26			in self._trusted_certificates.values())
27
28			def check_certificate(self, certificate):
29			return certificate in self
30
31			def get_user(self, certificate):
32			correct_users = [prospective_certificate['user'] for prospective_certificate in self._trusted_certificates.values()
33			if certificate == prospective_certificate['certificate']]
34			if len(correct_users) == 0:
35			return None
36			else:
37			return correct_users[0]
38

FreeOpcUa / opcua-asyncio

Pull Request — master (#209)

CertificateHandler.trust_certificate() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like