Page History

...

The article provides a method of authorization in gRPC channel with examples of Python code.

gRPC requests are generated on the basis of proto files.

Preparing the environment

Before you start, do the following:

이 문서는 Python 코드의 예와 함께 gRPC 채널에서 권한 부여 방법을 제공합니다.

gRPC 요청은 proto 파일을 기반으로 생성됩니다.

환경 준비

시작하기 전에 다음을 수행하십시오.

1. Python 인터프리터와 필요한 경우 IDE를 설치합니다.

2. pip를 사용하여 종속성(dependencies)을 설치합니.

3. Install the Python interpreter and, if necessary, the IDE.

4. Install the dependencies via pip:

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

Creating the proto classes

To create proto classes, do the following:

프로토 클래스 만들기

proto 클래스를 만들려면 다음을 수행합니다.

1. Technical Support에서 프로토 파일을 받으십시오.

2. 스크립트를 PY 파일로 저장합니다

3. Get proto files from Technical Support.

4. Save the script as a PY file.

   Expand
   title Click to expand...
   Code Block 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, AxxonSoft_protos_dir): proto_files_relative = get_proto_files_relpath(AxxonSoft_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(AxxonSoft_protos_dir, protos_dir): out = [] for root, dirs, files in os.walk(AxxonSoft_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' AxxonSoft_dir_name = 'AxxonSoft' current_dir = os.path.dirname(os.path.abspath(__file__)) protos_dir = os.path.join(current_dir, protos_dir_name) AxxonSoft_protos_dir = os.path.join(protos_dir, AxxonSoft_dir_name) bindings_package_dir = os.path.join(current_dir, AxxonSoft_dir_name) paths_print([ ('protos_dir', protos_dir), ('AxxonSoft_protos_dir', AxxonSoft_protos_dir), ('bindings_package_dir', bindings_package_dir), ]) clear_folder(bindings_package_dir) code = generate_bindings(protos_dir, AxxonSoft_protos_dir) if code == 0: generate_init_py_files(bindings_package_dir) return code if __name__ == '__main__': print('Axxon One NativeBL bindings generator.') print('To generate that bindings you need to copy to `grpc-proto-files` folder: ') print('1) `AxxonSoft` folder with AxxonSoft proto-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}')

   
   

5. Create the 스크립트 폴더에 grpc-proto-files folder in the script folder. Place the AxxonSoft and google folders in this folder along with their contents from the resulting archive with proto files.
6. Run the script.

As a result, the script folder will contain the AxxonSoft folder with proto classes, which will be used to work via the gRPC channel.

Authorization and first request

1.  폴더를 만듭니다. AxxonSoft 및 google 폴더를 proto 파일과 함께 결과 아카이브의 콘텐츠와 함께 이 폴더에 배치합니다.
2. 스크립트를 실행합니다.

결과적으로 스크립트 폴더에는 gRPC 채널을 통해 작업하는 데 사용되는 프로토 클래스가 있는 AxxonSoft 폴더가 포함됩니다.

승인 및 첫 번째 요청

gRPC 채널을 통해 요청을 보내려면 승인이 필요합니다. 이를 위해서는  To send requests through the gRPC channel, the authorization is required. To do this, use the Server certificate from the C:\ProgramData\AxxonSoft\Axxon One\Tickets folder.

You can be authorized only to the Server from the certificate.

 폴더에 있는 서버 인증서를 사용합니다.

인증서에서 서버에만 권한을 부여할 수 있습니다.

다음은 권한 부여의 예와 루트 단위를 가져오기 위한 ConfigurationService.ListUnits 요청의 예입니다Following is an example of authorization and an example of a ConfigurationService.ListUnits request to get the root unit.

import grpc

from OpenSSL import crypto
from grpc._channel import _InactiveRpcError

from AxxonSoft.bl.config.ConfigurationService_pb2 import ListUnitsRequest
from AxxonSoft.bl.config.ConfigurationService_pb2_grpc import ConfigurationServiceStub
from AxxonSoft.bl.auth.Authentication_pb2 import AuthenticateRequest
from 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\AxxonSoft\Axxon One\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}")

The get_authorized_channel function procedure takes the following as parameters:함수 프로시저는 다음을 매개변수로 사용합니다.

1. certificate_path − path to the certificate인증서 경로;
2. ip − Server 서버 IP address 주소 (디폴트 "127.0.0.1" by default);
3. port − gRPC API port (디폴트 20109 by default);
4. username − username 사용자 이름 (디폴트 "root" by default);
5. password − user password 사용자 암 (디폴트 "root" by default).