1

#!/bin/bash

     2

     3

set -x -e

     4

     5

echo "Instant encrypted OPC-UA server for test"

     6

     7

rm -f my_cert.pem my_cert.der my_private_key.pem

     8

     9

yes "" | openssl req -x509 -newkey rsa:2048 -keyout my_private_key.pem -out my_cert.pem \

    10

        -days 355 -nodes -addext "subjectAltName = URI:urn:example.org:FreeOpcUa:python-opcua"

    11

openssl x509 -outform der -in my_cert.pem -out my_cert.der

    12

    13

PROJECT_FILES_DIR=$BEREMIZPATH/tests/projects/opcua_browse_encrypted/project_files

    14

mkdir $PROJECT_FILES_DIR -p

    15

cp my_cert.der my_private_key.pem $PROJECT_FILES_DIR

    16

    17

echo "CERTS READY"

    18

    19

# Run server

    20

exec $BEREMIZPYTHONPATH - << EOF

    21

    22

import sys

    23

import os

    24

import time

    25

import asyncio

    26

    27

from asyncua import ua, Server

    28

from asyncua.server.users import User, UserRole

    29

    30

# Asyncua can't work without (over)simple shared cerificates/privkey.

    31

# No user is involved in that case, but asyncua needs it.

    32

# Over permessive User Manager hereafter helps cuting that corner.

    33

class AllAdminUserManager:

    34

    def get_user(self, iserver, username=None, password=None, certificate=None):

    35

        return User(role=UserRole.Admin)

    36

    37

async def main():

    38

    server = Server(user_manager=AllAdminUserManager())

    39

    host = os.environ.get("OPCUA_DEFAULT_HOST", "127.0.0.1")

    40

    endpoint = "opc.tcp://"+host+":4840/freeopcua/server/"

    41

    await server.init()

    42

    server.set_endpoint(endpoint)

    43

    44

    server.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt])

    45

    await server.load_certificate("my_cert.der")

    46

    await server.load_private_key("my_private_key.pem")

    47

    48

    uri = "http://beremiz.github.io"

    49

    idx = await server.register_namespace(uri)

    50

    51

    objects = server.get_objects_node()

    52

    53

    testobj = await objects.add_object(idx, "TestObject")

    54

    testvarout = await testobj.add_variable(idx, "TestOut", 1.2)

    55

    testvar = await testobj.add_variable(idx, "TestIn", 5.6)

    56

    await testvar.set_writable()

    57

    58

    await server.start()

    59

    60

    try:

    61

        while True:

    62

            await asyncio.sleep(1)

    63

            inval = await testvar.get_value()

    64

            print(inval)

    65

            await testvarout.set_value(inval*2)

    66

            sys.stdout.flush()

    67

    finally:

    68

        await server.stop()

    69

    70

asyncio.run(main())

    71

    72

EOF