В статье приводится способ авторизации в gRPC-канале с примерами кода на языке Python.

gRPC-запросы формируются на основе proto-файлов.

Подготовка окружения

Перед началом работы необходимо:

  1. Установить интерпретатор языка Python и при необходимости IDE.
  2. Через pip установить зависимости:

    pip>=21.1.2
    grpcio-tools>=1.38.0
    googleapis-common-protos
    pyOpenSSL==19.1.0


Создание proto-классов

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

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


    import 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}')



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

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

Авторизация и первый запрос

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

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

Ниже приводится пример авторизации и пример запроса ConfigurationService.ListUnits для получения корневого юнита.


import grpc

from OpenSSL import crypto
from grpc._channel import _InactiveRpcError

fromITV.bl.config.ConfigurationService_pb2 import ListUnitsRequest
fromITV.bl.config.ConfigurationService_pb2_grpc import ConfigurationServiceStub
fromITV.bl.auth.Authentication_pb2 import AuthenticateRequest
fromITV.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\ITV\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}")


Функция get_authorized_channel в качестве параметров принимает:

  1. certificate_path - путь к сертификату;
  2. ip - IP-адрес Сервера (по умолчанию "127.0.0.1");
  3. port - порт gRPC API (по умолчанию 20109);
  4. username - имя пользователя (по умолчанию "root");
  5. password - пароль пользователя (по умолчанию "root").

Импортируемые proto-классы из папкиITVбыли созданы на предыдущем шаге.