Page History

...

Для создания proto-классов необходимо:

1. Получить в службе технической поддержки proto-файлы.

2. Сохранить скрипт в виде py-файла.

   Expand Code Blockimport os import shutil import inspect import pkg_resources from grpc_tools.protoc import main as protoc POSIX_SEP = '/' def paths_print(items): print("Paths:") print("-"*80) for k, v in items: print("\t", k, "\t\t", v) print("-"*80) def clear_folder(output_dir): if os.path.exists(output_dir): shutil.rmtree(output_dir) def generate_bindings(protos_dir,ITV_protos_dir): proto_files_relative = get_proto_files_relpath(ITV_protos_dir, protos_dir) protoc_keys = [ f'-I{protos_dir}', '--python_out=.', '--grpc_python_out=.', ] protoc_args = protoc_keys + proto_files_relative is_protoc_patched = 'include_module_proto' in inspect.getfullargspec(protoc).args if not is_protoc_patched: protoc_args_patch_start = [inspect.getfile(protoc)] protoc_args_patch_end = [f'-I{pkg_resources.resource_filename("grpc_tools", "_proto")}'] else: protoc_args_patch_start = protoc_args_patch_end = [] print('Call of "protoc":') protoc_retcode = protoc(protoc_args_patch_start + protoc_args + protoc_args_patch_end) # print(f'\targs = {protoc_args}\n\tretcode = {protoc_retcode}\n') return protoc_retcode def generate_init_py_files(bindings_output_dir): def make_init_py_subtree(base_path): # print('\tprocess {!r}'.format(base_path)) make_init_py(base_path) for subdir in get_subdirs(base_path): make_init_py_subtree(subdir) def make_init_py(base_path): modules = get_py_modules(base_path) init_py_path = os.path.join(base_path, '__init__.py') with open(init_py_path, 'w') as init_py: init_py.write('# Generated AUTOMATICALLY by \"{}\".\n'.format(os.path.basename(__file__))) init_py.write('# DO NOT EDIT manually!\n\n') for m in modules: if '.Internal' not in m: init_py.write('from . import {!s}\n'.format(m)) def get_subdirs(base_path): _, subdirs, _ = next(os.walk(base_path)) return [os.path.abspath(os.path.join(base_path, s)) for s in subdirs] def get_py_modules(base_path): _, subdirs, files = next(os.walk(base_path)) file_modules = [f.rstrip('.py') for f in files if f.endswith('.py') and f != '__init__.py'] return subdirs + file_modules # print('Generate "__init__.py" files:') make_init_py_subtree(bindings_output_dir) def get_proto_files_relpath(ITV_protos_dir, protos_dir): out = [] for root, dirs, files in os.walk(ITV_protos_dir): for file in files: if file.endswith(".proto") and ".Internal" not in file: full_path = os.path.abspath(os.path.join(root, file)) rel_path = os.path.relpath(full_path, protos_dir) posix_rel_path = rel_path.replace(os.sep, POSIX_SEP) out.append(posix_rel_path) return out def run_generate(): protos_dir_name = 'grpc-proto-files' ITV_dir_name = 'ITV' current_dir = os.path.dirname(os.path.abspath(__file__)) protos_dir = os.path.join(current_dir, protos_dir_name) ITV_protos_dir = os.path.join(protos_dir,ITV_dir_name) bindings_package_dir = os.path.join(current_dir,ITV_dir_name) paths_print([ ('protos_dir', protos_dir), ('ITV_protos_dir',ITV_protos_dir), ('bindings_package_dir', bindings_package_dir), ]) clear_folder(bindings_package_dir) code = generate_bindings(protos_dir,ITV_protos_dir) if code == 0: generate_init_py_files(bindings_package_dir) return code if __name__ == '__main__': print('AxxonNext NativeBL bindings generator.') print('To generate that bindings you need to copy to `grpc-proto-files` folder: ') print('1) `ITV` folder withITVproto-files, ') print('2) `google` folder with Google common proto-files.') result = run_generate() if result == 0: print('Bindings generation was completed successfully') else: print(f'An error occurred while generating bindings: {result}')

   : script.py.

3. В папке со скриптом создать папку grpc-proto-files. В эту папку поместить папки ITVи axxonsoft и google вместе с их содержимым из полученного архива с proto-файлами.
4. Запустить скрипт.

В результате в папке со скриптом появится папкаITVс папка axxonsoft с proto-классами, которые будут использоваться для работы через gRPC-канал.

...

Для отправки запросов через gRPC-канал необходима авторизация. Для этого необходимо использовать сертификат Сервера из папки C:\ProgramData\ITVAxxonSoft\AxxonNext\Tickets.

Авторизация возможна только к Серверу из сертификата.

...

import grpc

from OpenSSL import crypto
from grpc._channel import _InactiveRpcError

fromITVfrom axxonsoft.bl.config.ConfigurationService_pb2 import ListUnitsRequest
fromITVfrom axxonsoft.bl.config.ConfigurationService_pb2_grpc import ConfigurationServiceStub
fromITVfrom axxonsoft.bl.auth.Authentication_pb2 import AuthenticateRequest
fromITVfrom axxonsoft.bl.auth.Authentication_pb2_grpc import AuthenticationServiceStub


def get_channel_credentials(cert_path):
    with open(cert_path, 'rb') as f:
        certificate = f.read()

    creds = grpc.ssl_channel_credentials(root_certificates=certificate)

    cert = crypto.load_certificate(crypto.FILETYPE_PEM, certificate)
    common_name = cert.get_subject().CN

    return creds, common_name


def get_ssl_channel(server, channel_creds, override_cn, auth_creds=None):
    channel_creds = grpc.composite_channel_credentials(channel_creds, auth_creds) if auth_creds else channel_creds
    return grpc.secure_channel(server, channel_creds, options=(('grpc.ssl_target_name_override', override_cn),))


def get_auth_credentials(simple_channel, username, password):
    client = AuthenticationServiceStub(simple_channel)
    auth_request = AuthenticateRequest(user_name=username, password=password)
    response = client.Authenticate(auth_request)
    auth_header = (response.token_name, response.token_value)
    auth_creds = grpc.metadata_call_credentials(
        lambda _, cb: cb([auth_header], None))
    return auth_creds


def get_authorized_channel(certificate_path, ip="127.0.0.1", port=20109, username="root", password="root"):
    server = f"{ip}:{port}"
    channel_creds, cert_common_name = get_channel_credentials(certificate_path)
    try:
        simple_channel = get_ssl_channel(server, channel_creds, cert_common_name)
        auth_creds = get_auth_credentials(simple_channel, username, password)
        return get_ssl_channel(server, channel_creds, cert_common_name, auth_creds)
    except _InactiveRpcError as ex:
        print(f"Unable to connect to server. Details:\n{ex.details()}")


if __name__ == '__main__':
    print('This script need to provide a path to the certificate')
    path = r"C:\ProgramData\ITVAxxonSoft\AxxonNext\Tickets\Node.crt"
    channel = get_authorized_channel(path)
    config_service = ConfigurationServiceStub(channel)
    request = ListUnitsRequest(unit_uids=["root"])
    response = config_service.ListUnits(request)
    print(f"Found {len(response.units)} units:\n{response.units}")

...

...