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-08 14:52 UTC

CertificateHandler.trust_certificate() A

↳ Parent: asyncua.crypto.certificate_handler

Complexity

Conditions

Size

Total Lines	9
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	8
nop	5
dl	0
loc	9
rs	10
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 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, user_role=UserRole.User):
13			certificate = await uacrypto.load_certificate(certificate_path, format)
14			if label is None:
15			label = certificate_path
16			user = User(role=user_role, name=label)
17			if label in self._trusted_certificates:
18			logging.warning(f"certificate with label {label} "
19			f"attempted to be added multiple times, only the last version will be kept.")
20			self._trusted_certificates[label] = {'certificate': uacrypto.der_from_x509(certificate), 'user':user}
21
22			def __contains__(self, certificate):
23			return any(certificate == prospective_cert['certificate']
24			for prospective_cert
25			in self._trusted_certificates.values())
26
27			def get_user(self, certificate):
28			correct_users = [prospective_certificate['user'] for prospective_certificate in self._trusted_certificates.values()
29			if certificate == prospective_certificate['certificate']]
30			if len(correct_users) == 0:
31			return None
32			else:
33			return correct_users[0]
34

FreeOpcUa / opcua-asyncio

Pull Request — master (#209)

CertificateHandler.trust_certificate() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like