edouard@3543: #!/bin/bash
edouard@3543: 
edouard@3820: set -x -e
edouard@3820: 
edouard@3676: echo "Instant encrypted OPC-UA server for test"
edouard@3676: 
edouard@3820: rm -f my_cert.pem my_cert.der my_private_key.pem
edouard@3820: 
edouard@3676: yes "" | openssl req -x509 -newkey rsa:2048 -keyout my_private_key.pem -out my_cert.pem \
edouard@3676:         -days 355 -nodes -addext "subjectAltName = URI:urn:example.org:FreeOpcUa:python-opcua"
edouard@3676: openssl x509 -outform der -in my_cert.pem -out my_cert.der
edouard@3676: 
edouard@3676: PROJECT_FILES_DIR=$BEREMIZPATH/tests/projects/opcua_browse_encrypted/project_files
edouard@3820: mkdir $PROJECT_FILES_DIR -p
edouard@3676: cp my_cert.der my_private_key.pem $PROJECT_FILES_DIR
edouard@3676: 
edouard@3676: echo "CERTS READY"
edouard@3543: 
edouard@3549: # Run server
edouard@3672: exec $BEREMIZPYTHONPATH - << EOF
edouard@3543: 
edouard@3549: import sys
edouard@3718: import os
edouard@3549: import time
edouard@3820: import asyncio
edouard@3549: 
edouard@3820: from asyncua import ua, Server
edouard@3820: from asyncua.server.users import User, UserRole
edouard@3549: 
edouard@3820: # Asyncua can't work without (over)simple shared cerificates/privkey.
edouard@3820: # No user is involved in that case, but asyncua needs it.
edouard@3820: # Over permessive User Manager hereafter helps cuting that corner.
edouard@3820: class AllAdminUserManager:
edouard@3820:     def get_user(self, iserver, username=None, password=None, certificate=None):
edouard@3820:         return User(role=UserRole.Admin)
edouard@3549: 
edouard@3820: async def main():
edouard@3820:     server = Server(user_manager=AllAdminUserManager())
edouard@3820:     host = os.environ.get("OPCUA_DEFAULT_HOST", "127.0.0.1")
edouard@3820:     endpoint = "opc.tcp://"+host+":4840/freeopcua/server/"
edouard@3820:     await server.init()
edouard@3820:     server.set_endpoint(endpoint)
edouard@3676: 
edouard@3820:     server.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt])
edouard@3820:     await server.load_certificate("my_cert.der")
edouard@3820:     await server.load_private_key("my_private_key.pem")
edouard@3549: 
edouard@3820:     uri = "http://beremiz.github.io"
edouard@3820:     idx = await server.register_namespace(uri)
edouard@3549: 
edouard@3820:     objects = server.get_objects_node()
edouard@3549: 
edouard@3820:     testobj = await objects.add_object(idx, "TestObject")
edouard@3820:     testvarout = await testobj.add_variable(idx, "TestOut", 1.2)
edouard@3820:     testvar = await testobj.add_variable(idx, "TestIn", 5.6)
edouard@3820:     await testvar.set_writable()
edouard@3549: 
edouard@3820:     await server.start()
edouard@3820: 
edouard@3820:     try:
edouard@3820:         while True:
edouard@3820:             await asyncio.sleep(1)
edouard@3820:             inval = await testvar.get_value()
edouard@3820:             print(inval)
edouard@3820:             await testvarout.set_value(inval*2)
edouard@3820:             sys.stdout.flush()
edouard@3820:     finally:
edouard@3820:         await server.stop()
edouard@3820: 
edouard@3820: asyncio.run(main())
edouard@3820: 
edouard@3549: EOF