vitor-aignosi
ecb055eefb
SIENTIAPDE-1973: Implement per-connector OPC client certificates and remove legacy read engine. This update overhauls certificate management, transitioning from a single platform certificate to individual client certificates defined in the slot configuration. The CertProvider now downloads certificates from MinIO based on certificate_id, applying dynamic security policies (mode, policy) from the slot. The legacy READ_ENGINE=legacy option and OpcManager have been entirely removed, standardizing on the ServerReader with strict slot v2 enforcement. Write command resolution (ServerConfigResolver) now consistently uses connectors joined with opc_certificates. The CERT_BUCKET environment variable is no longer used.
68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
import pytest
|
|
|
|
from ingestor.connection.certificates import (
|
|
client_certificate_id,
|
|
has_security,
|
|
resolve_certificate,
|
|
)
|
|
|
|
LOOKED_UP = {
|
|
'certificate_id': 1,
|
|
'application_uri': 'urn:sientia:opc-ingestor',
|
|
'cert': {'bucket': 'opc-certificates', 'object_name': 'client/p/cert.pem'},
|
|
'private_key': {'bucket': 'opc-certificates', 'object_name': 'client/p/key.pem'},
|
|
}
|
|
|
|
|
|
def _connector(certificates, certificate_id=1):
|
|
return {
|
|
'connector_id': 'kepserver',
|
|
'url': 'opc.tcp://kep:49320',
|
|
'security': {
|
|
'mode': 'SignAndEncrypt',
|
|
'policy': 'Basic256Sha256',
|
|
'client_certificate_id': certificate_id,
|
|
},
|
|
'certificates': certificates,
|
|
}
|
|
|
|
|
|
def test_resolve_certificate_keeps_only_the_slot_fields():
|
|
looked_up = {
|
|
**LOOKED_UP,
|
|
'thumbprint_sha1': 'ab',
|
|
'cert': {**LOOKED_UP['cert'], 'sha256': None},
|
|
}
|
|
|
|
assert resolve_certificate(_connector([looked_up])) == LOOKED_UP
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'certificates',
|
|
[
|
|
pytest.param(None, id='no-lookup'),
|
|
pytest.param([], id='not-found'),
|
|
pytest.param([{**LOOKED_UP, 'application_uri': ''}], id='no-application-uri'),
|
|
pytest.param([{**LOOKED_UP, 'cert': None}], id='no-cert'),
|
|
pytest.param([{**LOOKED_UP, 'private_key': {'object_name': 'k'}}], id='key-no-bucket'),
|
|
pytest.param(['not a document'], id='not-a-document'),
|
|
],
|
|
)
|
|
def test_resolve_certificate_without_a_usable_certificate(certificates):
|
|
assert resolve_certificate(_connector(certificates)) is None
|
|
|
|
|
|
def test_client_certificate_id():
|
|
assert client_certificate_id(_connector([])) == 1
|
|
assert client_certificate_id(_connector([], certificate_id=None)) is None
|
|
assert client_certificate_id({'connector_id': 'x'}) is None
|
|
assert client_certificate_id({'security': None}) is None
|
|
|
|
|
|
def test_has_security():
|
|
assert has_security(_connector([]))
|
|
assert has_security({'security': {'mode': 'None', 'policy': 'None'}})
|
|
assert not has_security({'security': {'mode': 'Sign'}})
|
|
assert not has_security({'security': None})
|
|
assert not has_security({})
|