Met difficulties during enabling encryption between server and client

[Sm1Ling]

Hello everyone!

Currently trying to set OPC Server-Client connection with encryption. I already have studied opcua-asyncio/examples/client-with-encryption.py, opcua-asyncio/examples/server-with-encryption.py and opcua-asyncio/examples/generate_certificates.py

However, I slightly changed code (I do not want to use predefined certificates and private keys from examples)

1. I generated Server private key, CA private key and CA certificate, then generated server's CSR for CA. With the help of CA I signed and trusted Server certificate
2. Generated Client private key and self-signed certificate
3. • Started Server app with URI equal to one I used in CSR (I did not find any other URI connected with server during keys/certificates generation),
   • Used Client self-signed certificate in Server's CertificateUserManager (as the link to the client, I believe?)
   • Loaded server's private key and signed certificate to enable encryption
4. • Started Client app with URI equal to one I used in selfsigned certificate
   • Set security to the client with Client's private key, Client's certificate, Server's signed certificate
5. Server writes the data. Client tries to connect and read it. Simple
6. I get ERROR:main:The certificate is not trusted.(BadCertificateUntrusted)

I believe, that the cause lies in certificates and URI's theme. But I can not understand, where and why.
Could you help me to sort out the problem? I believe that will become usefull hint for future users

Server app

url = "opc.tcp://127.0.0.1:4840/freeopcua/openserver/"

working_iterations = 20

cert_base = Path(__file__).parent.parent / "cert"
server_cert = cert_base / "certs" / "server_cert.der" # signed by CA via CSR
server_private_key = cert_base / "private" / "server_private_key.pem"
client_cert = cert_base / "certs" / "client_selfsigned_cert.der"
 

USE_TRUST_STORE = False


async def main():

    HOSTNAME = socket.gethostname()
    server_app_uri = f"urn:{HOSTNAME}:my:myserver"
    print(server_app_uri)

    cert_user_manager = CertificateUserManager()
    # Undefined place
    await cert_user_manager.add_user(client_cert, name="test_user")

    server = Server(user_manager=cert_user_manager)

    await server.init()

    await server.set_application_uri(server_app_uri)
    server.set_endpoint(url)
    server.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt])

    # load server certificate and private key. This enables endpoints
    # with signing and encryption.
    await server.load_certificate(str(server_cert))
    await server.load_private_key(str(server_private_key))

    if USE_TRUST_STORE:
        trust_store = TrustStore([cert_base / "certs" / "trusted"], [])
        await trust_store.load()
        validator = CertificateValidator(
            options=CertificateValidatorOptions.TRUSTED_VALIDATION | CertificateValidatorOptions.PEER_CLIENT,
            trust_store=trust_store,
        )
    else:
        validator = CertificateValidator(
            options=CertificateValidatorOptions.EXT_VALIDATION | CertificateValidatorOptions.PEER_CLIENT
        )
    server.set_certificate_validator(validator)

    idx = 0

    # populating our address space
    myobj = await server.nodes.objects.add_object(idx, "MyObject")
    myvar = await myobj.add_variable(idx, "MyVariable", 0.0)
    await myvar.set_writable()  # Set MyVariable to be writable by clients

    # starting!

    async with server:
        for _ in range(working_iterations):
            await asyncio.sleep(3)
            current_val = await myvar.get_value()
            count = current_val + 0.1
            
            print(f"Set var value of {myvar} to {count}")
            print(f"Object is {myobj}")
            await myvar.write_value(count)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

Client app

logging.basicConfig(level=logging.INFO)
_logger = logging.getLogger(__name__)

USE_TRUST_STORE = True

url = "opc.tcp://127.0.0.1:4840/freeopcua/openserver/"

cert_base = Path(__file__).parent.parent / "cert"
client_cert = Path(cert_base / "certs" / "client_selfsigned_cert.der")
server_cert = Path(cert_base / "certs" / "server_cert.der")
client_private_key = Path(cert_base / "private" / "client_private_key.pem")


async def task(loop):
    HOSTNAME = socket.gethostname()
    server_app_uri = f"urn:{HOSTNAME}:my:myselfsignedclient"
    print(server_app_uri)

    client = Client(url=url)
    client.application_uri = server_app_uri
    await client.set_security(
        SecurityPolicyBasic256Sha256,
        certificate=str(client_cert),
        private_key=str(client_private_key),
        server_certificate=server_cert,
    )

    if USE_TRUST_STORE:
        trust_store = TrustStore([Path("examples") / "certificates" / "trusted" / "certs"], [])
        await trust_store.load()
        validator = CertificateValidator(
            CertificateValidatorOptions.TRUSTED_VALIDATION | CertificateValidatorOptions.PEER_SERVER, trust_store
        )
    else:
        validator = CertificateValidator(
            CertificateValidatorOptions.EXT_VALIDATION | CertificateValidatorOptions.PEER_SERVER
        )
    client.certificate_validator = validator
    try:
        async with client:
            objects = client.nodes.objects
            child = await objects.get_child(["0:MyObject", "0:MyVariable"])
            print(await child.get_value())
            await child.set_value(42)
            print(await child.get_value())
    except ua.UaError as exp:
        _logger.error(exp)


def main():
    loop = asyncio.get_event_loop()
    loop.set_debug(True)
    loop.run_until_complete(task(loop))
    loop.close()


if __name__ == "__main__":
    main()

Script for password generation

HOSTNAME: str = socket.gethostname()

# used for subject common part
NAMES: Dict[str, str] = {
    "countryName": "UZ",
    "stateOrProvinceName": "TAS",
    "localityName": "MyLoc",
    "organizationName": "MyComp",
}

CLIENT_SERVER_USE = [ExtendedKeyUsageOID.CLIENT_AUTH, ExtendedKeyUsageOID.SERVER_AUTH]

# setup the paths for the certs, keys and csr
base = Path("cert")
base_csr: Path = base / "csr"
base_private: Path = base / "private"
base_certs: Path = base / "certs"
base_csr.mkdir(parents=True, exist_ok=True)
base_private.mkdir(parents=True, exist_ok=True)
base_certs.mkdir(parents=True, exist_ok=True)

server_key_name = "server_private_key.pem"
server_cert_selfsigned_name = "server_selfsigned_cert.der"
client_key_name = "client_private_key.pem"
client_cert_selfsigned_name = "client_selfsigned_cert.der"
server_cert_name = "server_cert.der"
server_csr_name = "server_csr.csr"
ca_key_name = "ca_application.pem"
ca_cert_name = "ca_application.der"


def save_private_key(key_path: Path):
    key: RSAPrivateKey = generate_private_key()
    key_path.write_bytes(dump_private_key_as_pem(key))


async def generate_self_signed_certificate(key_path: Path, cert_path: Path, appname: str):
    name = f"myselfsigned{appname}"
    subject_alt_names: List[x509.GeneralName] = [
        # Undefined place
        x509.UniformResourceIdentifier(f"urn:{HOSTNAME}:my:{name}"),
        x509.DNSName(f"{HOSTNAME}"),
    ]

    key = await load_private_key(key_path)

    cert: x509.Certificate = generate_self_signed_app_certificate(
        # Undefined place
        key, f"{name}@{HOSTNAME}", NAMES, subject_alt_names, extended=CLIENT_SERVER_USE
    )

    cert_path.write_bytes(cert.public_bytes(encoding=Encoding.DER))

def generate_applicationgroup_ca():
    subject_alt_names: List[x509.GeneralName] = [
        # Undefined place
        x509.UniformResourceIdentifier(f"urn:{HOSTNAME}:my:myca"),
        x509.DNSName(f"{HOSTNAME}"),
    ]

    key: RSAPrivateKey = generate_private_key()
    cert: x509.Certificate = generate_self_signed_app_certificate(
        key, "Application CA", NAMES, subject_alt_names, extended=[]
    )

    key_file = base_private / ca_key_name
    cert_file = base_certs / ca_cert_name

    key_file.write_bytes(dump_private_key_as_pem(key))
    cert_file.write_bytes(cert.public_bytes(encoding=Encoding.DER))


async def generate_csr():
    subject_alt_names: List[x509.GeneralName] = [
        x509.UniformResourceIdentifier(f"urn:{HOSTNAME}:my:myserver"),
        x509.DNSName(f"{HOSTNAME}"),
    ]

    key = await load_private_key(base_private / server_key_name)
    csr: x509.CertificateSigningRequest = generate_app_certificate_signing_request(
        key, f"myserver@{HOSTNAME}", NAMES, subject_alt_names, extended=CLIENT_SERVER_USE
    )

    csr_file = base_csr / server_csr_name
    csr_file.write_bytes(csr.public_bytes(encoding=Encoding.PEM))


async def sign_csr():
    issuer = await load_certificate(base_certs / ca_cert_name)
    key_ca = await load_private_key(base_private / ca_key_name )
    csr_file: Path = base_csr / server_csr_name
    csr = x509.load_pem_x509_csr(csr_file.read_bytes())

    cert: x509.Certificate = sign_certificate_request(csr, issuer, key_ca, days=30)
    (base_certs / server_cert_name).write_bytes(cert.public_bytes(encoding=Encoding.DER))


async def main():
    # create key and reuse it for self_signed and generate_csr
    save_private_key(base_private / server_key_name)
    save_private_key(base_private / client_key_name)

    # generate self signed certificate for myserver-selfsigned
    await generate_self_signed_certificate(
        base_private / server_key_name,
        base_certs / server_cert_selfsigned_name,
        "server"
    )
    await generate_self_signed_certificate(
        base_private / client_key_name,
        base_certs / client_cert_selfsigned_name,
        "client"
    )

    # generate certificate signing request and sign it with the ca for myserver
    generate_applicationgroup_ca()
    await generate_csr()
    await sign_csr()


if __name__ == "__main__":
    asyncio.run(main())

[Sm1Ling]

Found out that it throws error if USE_TRUST_STORE == False in Server and USE_TRUST_STORE ==True in Client

After I set USE_TRUST_STORE = False in Client, the Client got ability to connect to server. However, it still blocks ability for client to write

ERROR:__main__:User does not have permission to perform the requested operation.(BadUserAccessDenied)

I will try to find out how to provide Client access for writing

[Sm1Ling]

Things to test:

• Will it work with USE_TRUST_STORE ==True both in Server and Client
• Will it work without CA certification. Only with server and client self-signed certificates
• Try to provide Client admin rights
• Find the correct way how to provide ability to write values from client

[Sm1Ling]

USE_TRUST_STORE ==True did not work because... trust store folder is empty. Now I will try to understand, what certificates should be in there

[Sm1Ling]

Moved both server's CA signed certificate and self-signed certificates to Trusted folder. Changed paths in application. Set USE_TRUST_STORE =True. Got ERROR:__main__:The certificate is not trusted.(BadCertificateUntrusted) So, this is not something the applications expected. Currently I freeze attempts on handling the Trust problem

[Sm1Ling]

Option with selfsigned certificates works. I changed CA signed server certificate to selfsigned, updated server app uri and connection was created. Howerver, still without writing

[Sm1Ling]

Handled writing restrictions. Everything we have to do is to change in Server application

    cert_user_manager = CertificateUserManager()
    await cert_user_manager.add_user(client_cert, name="test_user")

to

    cert_user_manager = CertificateUserManager()
    await cert_user_manager.add_admin(client_cert, name="test_user")

Probably, only users with admin privileges can write to the server